From 046433ee9552a8f54295fad6cf47f3aa975162c6 Mon Sep 17 00:00:00 2001 From: Shaurya K Sharma Date: Sat, 29 Aug 2026 22:28:33 +0100 Subject: [PATCH 1/4] refactor(scale): claim analysis jobs through an explicit control plane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Analysis jobs were already durable, leased, cancellable and recoverable, but "who owns a job" was expressible only as private methods of AnalysisWorker: the claim compare-and-swap, the lease-renewal compare-and-swap, the expired-lease scan and the ownership predicate all lived inside the executor. A second worker process could not participate in the queue without importing the class that runs the extraction pipeline. Introduce app/workers/control_plane.py as that boundary. AnalysisControlPlane is the protocol; DatabaseAnalysisControlPlane is the v1 implementation, and it is the existing analysis_jobs table rather than a new broker — the durable row is already the authority for identity, attempt budget, cancellation and lease expiry, so a queue service would only add a second, weaker source of truth. JobLease is a value, not a handle: ownership is re-proved by a guard on every mutation, so a stale lease can never write to a reclaimed job. LeaseRenewal distinguishes renewed / lost / deferred, making the previously unnamed SQLite single-writer case explicit instead of an inline exception filter. AnalysisWorker keeps execution only: running a job it already owns through the ri.v1 pipeline and deciding its terminal transition. Every statement moved across is unchanged, so the claim/lease/cancellation contract is identical; _claim stays as a row-shaped delegation for existing callers. Refs #324, #210 --- apps/backend/app/workers/analysis_worker.py | 268 +++++------- apps/backend/app/workers/control_plane.py | 428 ++++++++++++++++++++ apps/backend/tests/test_analysis_worker.py | 7 +- 3 files changed, 533 insertions(+), 170 deletions(-) create mode 100644 apps/backend/app/workers/control_plane.py diff --git a/apps/backend/app/workers/analysis_worker.py b/apps/backend/app/workers/analysis_worker.py index 78995bb0..a3a455c2 100644 --- a/apps/backend/app/workers/analysis_worker.py +++ b/apps/backend/app/workers/analysis_worker.py @@ -4,11 +4,19 @@ full analysis off the request path through the evidence-backed extraction pipeline that seals the repository's authoritative ``ri.v1`` snapshot. -``run_once`` is the primary unit both the background loop in ``app.main`` and the -test-suite drive; it is synchronous and deterministic. It claims at most one job -with a portable compare-and-swap (no ``SELECT ... FOR UPDATE SKIP LOCKED``), runs -the stages with cooperative cancellation checks within and between them, and applies a -bounded exponential backoff on failure before finally marking the job ``failed``. +``run_once`` is the primary unit both ``AnalysisWorkerRunner`` and the test suite +drive; it is synchronous and deterministic. It claims at most one job *through +the control plane* (``app.workers.control_plane``), runs the stages with +cooperative cancellation checks within and between them, and applies a bounded +exponential backoff on failure before finally marking the job ``failed``. + +Boundary note (#324): this class no longer decides **who owns a job**. Claiming, +lease renewal, expiry, reclaim and every ownership guard are +``AnalysisControlPlane`` calls, and the loop that drives ``run_once`` lives in +``app.workers.runner``. What remains here is execution: turning a job this +worker *already owns* into a sealed ``ri.v1`` snapshot, and deciding its +terminal transition. Keep it that way -- queue policy belongs on the other side +of the boundary, and repository analysis belongs on this side of it. Transactional note (deliberate design, not a gap to "fix"): ``SnapshotStore.seal`` owns its own commit boundary (``snapshot_store.py`` ``_commit_transition``) and @@ -33,7 +41,6 @@ from datetime import UTC, datetime, timedelta from pathlib import Path -from sqlalchemy import func, or_, select, update from sqlalchemy.orm import Session from app.analysis.resource_budget import ( @@ -67,6 +74,12 @@ ANALYSIS_PRODUCER_VERSION_SET, ANALYSIS_SCHEMA_VERSION, ) +from app.workers.control_plane import ( + AnalysisControlPlane, + DatabaseAnalysisControlPlane, + JobLease, + lease_expired, +) _MAX_ERROR_MESSAGE = 1024 _ERROR_CODE = "RI-JOB-FAILED" @@ -103,6 +116,7 @@ class _StageContext: session: Session job: AnalysisJob record: RepositoryRecord | None + lease: JobLease | None = None store: SnapshotStore | None = None snapshot: RiSnapshot | None = None reused: bool = False @@ -127,10 +141,18 @@ def __init__( monotonic: Callable[[], float] | None = None, clock: Callable[[], datetime] = lambda: datetime.now(UTC), heartbeat_interval_seconds: float | None = None, + control_plane: AnalysisControlPlane | None = None, ) -> None: self.session_factory = session_factory self.worker_id = worker_id self.lease_seconds = lease_seconds + # The queue this worker draws from. Defaulting to the durable + # ``analysis_jobs`` table keeps every existing caller unchanged while + # making the boundary an argument rather than an assumption. + self.control_plane: AnalysisControlPlane = control_plane or DatabaseAnalysisControlPlane( + lease_seconds=lease_seconds, + clock=clock, + ) self.max_source_bytes = max_source_bytes self.max_repository_source_bytes = max_repository_source_bytes self.max_process_rss_bytes = max_process_rss_bytes @@ -152,12 +174,15 @@ def run_once(self) -> bool: session = self.session_factory() try: - job = self._claim(session) + lease = self.control_plane.claim(session, worker_id=self.worker_id) + if lease is None: + return False + job = session.get(AnalysisJob, lease.job_id) if job is None: return False # Draining the stage generator runs the whole job to a terminal # state; tests can instead step the generator to interleave a cancel. - for _ in self._execute_stages(session, job): + for _ in self._execute_stages(session, job, lease=lease): pass return True finally: @@ -181,17 +206,7 @@ def sweep_stale(self) -> int: session = self.session_factory() try: now = self._clock() - stale_ids = tuple( - session.scalars( - select(AnalysisJob.id) - .where( - AnalysisJob.status == "running", - AnalysisJob.lease_expires_at.is_not(None), - AnalysisJob.lease_expires_at < now, - ) - .order_by(AnalysisJob.lease_expires_at, AnalysisJob.created_at) - ) - ) + stale_ids = self.control_plane.expired_job_ids(session, now=now) reclaimed = 0 for job_id in stale_ids: # Re-read each row so another sweeper that already reconciled it @@ -202,7 +217,7 @@ def sweep_stale(self) -> int: job is None or job.status != "running" or job.lease_expires_at is None - or not self._lease_expired(job.lease_expires_at, now) + or not lease_expired(job.lease_expires_at, now) ): continue if self._reconcile_stale(session, job, now): @@ -214,61 +229,40 @@ def sweep_stale(self) -> int: # -- claim --------------------------------------------------------------- def _claim(self, session: Session) -> AnalysisJob | None: - """Atomically claim the oldest eligible queued job, or return None. + """Claim one job through the control plane, as a row. - The claim is a portable compare-and-swap rather than - ``SELECT ... FOR UPDATE SKIP LOCKED``: pick the oldest eligible id, then - ``UPDATE ... WHERE id = :id AND status='queued'``. The ``status='queued'`` - predicate is the atomic guard — two workers racing for the same row see - exactly one non-zero ``rowcount``; the loser gets ``None`` and retries. + The claim/lease contract itself lives in + ``app.workers.control_plane``; this is the row-shaped convenience the + execution paths and tests use. """ - now = self._clock() - candidate_id = session.scalar( - select(AnalysisJob.id) - .where( - AnalysisJob.status == "queued", - or_(AnalysisJob.next_attempt_at.is_(None), AnalysisJob.next_attempt_at <= now), - ) - .order_by(AnalysisJob.created_at) - .limit(1) - ) - if candidate_id is None: - return None - result = session.execute( - update(AnalysisJob) - .where(AnalysisJob.id == candidate_id, AnalysisJob.status == "queued") - .values( - status="running", - worker_id=self.worker_id, - lease_expires_at=now + timedelta(seconds=self.lease_seconds), - started_at=func.coalesce(AnalysisJob.started_at, now), - attempt=AnalysisJob.attempt + 1, - next_attempt_at=None, - updated_at=now, - ) - ) - session.commit() - if result.rowcount == 0: - # Another worker won the compare-and-swap for this row. + lease = self.control_plane.claim(session, worker_id=self.worker_id) + if lease is None: return None - return session.get(AnalysisJob, candidate_id) + return session.get(AnalysisJob, lease.job_id) # -- stage pipeline ------------------------------------------------------ - def _execute_stages(self, session: Session, job: AnalysisJob) -> Iterator[_StageContext]: + def _execute_stages( + self, + session: Session, + job: AnalysisJob, + *, + lease: JobLease | None = None, + ) -> Iterator[_StageContext]: """Run the job's stages, yielding at each boundary. - Yielding after every stage gives the background loop a plain drain and - gives tests a seam to set ``cancel_requested`` between two stages. The - cancel flag is re-read from the row before each stage so a concurrent - ``AnalysisJobService.cancel`` is observed cooperatively. + Yielding after every stage gives the runner loop a plain drain and gives + tests a seam to set ``cancel_requested`` between two stages. The cancel + flag is re-read from the row through the control plane before each stage + so a concurrent ``AnalysisJobService.cancel`` is observed cooperatively. """ ctx = _StageContext( session=session, job=job, record=session.get(RepositoryRecord, job.repository_id), + lease=lease, ) stages = ( ("extracting-modules", 35, self._stage_open_snapshot), @@ -612,31 +606,13 @@ def _fail_resource_exceeded(self, ctx: _StageContext, exc: AnalysisResourceExcee def _reconcile_stale(self, session: Session, job: AnalysisJob, now: datetime) -> bool: """Atomically claim and reconcile one expired running job.""" - stale_worker_id = job.worker_id - stale_lease_expires_at = job.lease_expires_at - result = session.execute( - update(AnalysisJob) - .where( - AnalysisJob.id == job.id, - AnalysisJob.status == "running", - AnalysisJob.worker_id == stale_worker_id, - AnalysisJob.lease_expires_at == stale_lease_expires_at, - AnalysisJob.lease_expires_at < now, - ) - .values( - worker_id=self.worker_id, - lease_expires_at=now + timedelta(seconds=self.lease_seconds), - updated_at=now, - ) - .execution_options(synchronize_session="fetch") - ) - if result.rowcount == 0: - session.rollback() + lease = self.control_plane.reclaim(session, job, worker_id=self.worker_id) + if lease is None: return False record = session.get(RepositoryRecord, job.repository_id) snapshot = session.get(RiSnapshot, job.snapshot_id) if job.snapshot_id else None - ctx = _StageContext(session=session, job=job, record=record) + ctx = _StageContext(session=session, job=job, record=record, lease=lease) if job.cancel_requested: self._cancel(ctx) @@ -732,18 +708,14 @@ def _update_owned_job( """Update a running job only while this worker still owns it.""" job_id = ctx.job.id - ownership = [ - AnalysisJob.id == job_id, - AnalysisJob.worker_id == self.worker_id, - AnalysisJob.status == "running", - ] - if require_cancel_not_requested: - ownership.append(AnalysisJob.cancel_requested.is_(False)) - with ctx.session.no_autoflush: - result = ctx.session.execute( - update(AnalysisJob).where(*ownership).values(**values).execution_options(synchronize_session="fetch") - ) - if result.rowcount == 0: + held = self.control_plane.update_owned( + ctx.session, + job_id=job_id, + worker_id=self.worker_id, + values=values, + require_cancel_not_requested=require_cancel_not_requested, + ) + if not held: ctx.session.rollback() raise _LeaseLostError(job_id) @@ -770,7 +742,7 @@ def _lock_owned_job(self, ctx: _StageContext, *, require_cancel_not_requested: b raise def _cancel_requested(self, ctx: _StageContext) -> bool: - return bool(ctx.session.scalar(select(AnalysisJob.cancel_requested).where(AnalysisJob.id == ctx.job.id))) + return self.control_plane.cancel_requested(ctx.session, ctx.job.id) def _fail_open_snapshot(self, ctx: _StageContext, *, code: str) -> None: """Mark an opened-but-unsealed snapshot ``failed`` (no-op if reused/sealed).""" @@ -787,18 +759,38 @@ def _log_lease_lost(self, job_id: str) -> None: extra={"job_id": job_id, "worker_id": self.worker_id}, ) + def _lease_for(self, ctx: _StageContext) -> JobLease: + """This worker's ownership token for ``ctx``'s job. + + Paths that did not claim the job themselves — a reconciling sweep, or a + test driving a single stage — still own the row by ``worker_id``, which + is exactly what every control-plane guard tests. Rebuilding the token + from the row therefore asserts the same ownership a claim would have, + and never more than it. + """ + + if ctx.lease is not None: + return ctx.lease + return JobLease( + job_id=ctx.job.id, + worker_id=self.worker_id, + expires_at=ctx.job.lease_expires_at or self._clock(), + attempt=ctx.job.attempt, + ) + @contextmanager def _heartbeat(self, ctx: _StageContext) -> Iterator[_HeartbeatState]: """Renew one running job from an independent session during a stage.""" state = _HeartbeatState() - self._heartbeat_once(ctx.job.id, state) + lease = self._lease_for(ctx) + self._heartbeat_once(lease, state) def _run() -> None: while not state.stop.wait(self._heartbeat_interval_seconds): if self._shutdown.is_set(): return - self._heartbeat_once(ctx.job.id, state) + self._heartbeat_once(lease, state) if state.ownership_lost.is_set() or state.cancel_requested.is_set() or state.failure: return @@ -823,73 +815,27 @@ def _run() -> None: extra={"job_id": ctx.job.id, "worker_id": self.worker_id}, ) - def _heartbeat_once(self, job_id: str, state: _HeartbeatState) -> None: - """Atomically renew ownership and return the cancellation flag.""" - - session = self.session_factory() - sqlite_connection = None - sqlite_busy_timeout = None + def _heartbeat_once(self, lease: JobLease, state: _HeartbeatState) -> None: + """Pulse one lease renewal on an independent session. - def _restore_sqlite_timeout() -> None: - nonlocal sqlite_connection - if sqlite_connection is None or sqlite_busy_timeout is None: - return - try: - cursor = sqlite_connection.cursor() - cursor.execute(f"PRAGMA busy_timeout = {sqlite_busy_timeout}") - cursor.close() - except Exception: # noqa: BLE001 - discard a modified connection - session.invalidate() - finally: - sqlite_connection = None + The renewal itself — including the atomic cancellation read-back — is a + control-plane call; this method only owns the session and translates the + outcome into the thread-safe signals the stage loop watches. A + ``deferred`` outcome leaves every signal clear on purpose: ownership is + unchanged, so the stage must neither abandon its work nor treat the + skipped pulse as a failure. + """ + session = self.session_factory() try: - if session.bind is not None and session.bind.dialect.name == "sqlite": - connection = session.connection() - sqlite_connection = connection.connection.driver_connection - cursor = sqlite_connection.cursor() - sqlite_busy_timeout = int(cursor.execute("PRAGMA busy_timeout").fetchone()[0]) - # SQLite serializes all writers. If the stage already owns the - # database write lock, a sweeper cannot reclaim the job either, - # so the heartbeat must not block stage cleanup behind that lock. - cursor.execute("PRAGMA busy_timeout = 0") - cursor.close() - now = self._clock() - row = session.execute( - update(AnalysisJob) - .where( - AnalysisJob.id == job_id, - AnalysisJob.status == "running", - AnalysisJob.worker_id == self.worker_id, - ) - .values( - lease_expires_at=now + timedelta(seconds=self.lease_seconds), - updated_at=now, - ) - .returning(AnalysisJob.cancel_requested) - .execution_options(synchronize_session=False) - ).first() - if row is None: - _restore_sqlite_timeout() - session.rollback() + renewal = self.control_plane.renew(session, lease) + if renewal.lost: state.ownership_lost.set() - return - _restore_sqlite_timeout() - session.commit() - if row[0]: + elif renewal.cancel_requested: state.cancel_requested.set() except Exception as exc: # noqa: BLE001 - surfaced to the owning worker - _restore_sqlite_timeout() - session.rollback() - # A SQLite writer prevents every other SQLite writer, including a - # stale sweeper. Skipping that transient pulse avoids a false retry; - # PostgreSQL heartbeats remain fully independent and guarded. - if session.bind is not None and session.bind.dialect.name == "sqlite" and "locked" in str(exc).lower(): - logger.debug("SQLite analysis heartbeat skipped while the stage held the write lock") - else: - state.failure = exc + state.failure = exc finally: - _restore_sqlite_timeout() session.close() @staticmethod @@ -1051,16 +997,6 @@ def _backoff(attempt: int) -> int: return min(2**attempt, 60) - @staticmethod - def _lease_expired(lease_expires_at: datetime, now: datetime) -> bool: - """Compare SQLite-naive and timezone-aware persisted timestamps safely.""" - - if lease_expires_at.tzinfo is None and now.tzinfo is not None: - lease_expires_at = lease_expires_at.replace(tzinfo=UTC) - elif lease_expires_at.tzinfo is not None and now.tzinfo is None: - now = now.replace(tzinfo=UTC) - return lease_expires_at < now - @staticmethod def _error_message(exc: Exception) -> str: message = str(exc) or exc.__class__.__name__ diff --git a/apps/backend/app/workers/control_plane.py b/apps/backend/app/workers/control_plane.py new file mode 100644 index 00000000..62674e07 --- /dev/null +++ b/apps/backend/app/workers/control_plane.py @@ -0,0 +1,428 @@ +"""The analysis queue and control-plane boundary (#324). + +``analysis_jobs`` *is* the queue. This module is the explicit boundary around +that fact: everything that decides **which work is eligible** and **who owns +it** lives here, and nothing here knows how a repository is analysed. + +The split this module introduces +-------------------------------- + +Before #324 the claim compare-and-swap, the lease-renewal compare-and-swap, the +expired-lease scan and the ownership predicate were private methods of +``AnalysisWorker``, and the polling/sweep policy that drove them lived in +``app.main``. Ownership was therefore expressible only as "whatever the executor +happens to do", and a second worker process would have had to import the +executor to participate in the queue at all. + +``AnalysisControlPlane`` is now the seam: + +* **control plane** -- eligibility, claiming, lease renewal, expiry, reclaim, + ownership-guarded mutation, and observing a cancellation request; +* **executor** (``AnalysisWorker``) -- running a *claimed* job through the + Repository Intelligence pipeline and deciding its terminal transition. + +Why the database and not Redis/Celery +------------------------------------- + +The durable ``analysis_jobs`` row is already the authority for job identity, +attempt budget, cancellation and lease expiry, and it is already crash-safe by +idempotent reconciliation (see the ``analysis_worker`` module docstring). A +broker would add a second, weaker source of truth that still could not be +trusted over the row, plus a runtime dependency the deployment does not +otherwise need. The protocol below is the abstraction that makes a different +backing store possible later without another redesign of the claim/lease +contract; it is deliberately not that store. + +Portability +----------- + +Every mutation is a portable compare-and-swap (``UPDATE ... WHERE ``) +rather than ``SELECT ... FOR UPDATE SKIP LOCKED``, so SQLite development and +PostgreSQL deployment share one code path and one contract. PostgreSQL is the +deployment authority; the one dialect-specific concession is the SQLite +busy-timeout handling in :meth:`DatabaseAnalysisControlPlane.renew`, which is +documented at its site. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any, Literal, Protocol, cast + +from sqlalchemy import CursorResult, func, or_, select, update +from sqlalchemy.orm import Session +from sqlalchemy.sql.base import Executable + +from app.models.analysis_job import AnalysisJob + +logger = logging.getLogger(__name__) + +RenewalOutcome = Literal["renewed", "lost", "deferred"] + + +def lease_expired(lease_expires_at: datetime, now: datetime) -> bool: + """Compare SQLite-naive and timezone-aware persisted timestamps safely.""" + + if lease_expires_at.tzinfo is None and now.tzinfo is not None: + lease_expires_at = lease_expires_at.replace(tzinfo=UTC) + elif lease_expires_at.tzinfo is not None and now.tzinfo is None: + now = now.replace(tzinfo=UTC) + return lease_expires_at < now + + +@dataclass(frozen=True, slots=True) +class JobLease: + """Proof that ``worker_id`` owns ``job_id`` until ``expires_at``. + + A lease is a *value*, not a handle: holding one asserts nothing on its own. + Ownership is re-proved by the guard on every mutation + (:meth:`AnalysisControlPlane.update_owned`), so a stale lease object can + never be used to write to a job another worker has since reclaimed. + """ + + job_id: str + worker_id: str + expires_at: datetime + attempt: int + + +@dataclass(frozen=True, slots=True) +class LeaseRenewal: + """The outcome of one renewal attempt. + + ``deferred`` is neither a failure nor a loss: SQLite serialises writers, so + a renewal can be impossible *while this same worker's stage holds the write + lock*. Ownership is unchanged in that case -- and a sweeper is equally + locked out -- so the pulse is skipped rather than treated as a lost lease. + """ + + outcome: RenewalOutcome + cancel_requested: bool = False + expires_at: datetime | None = None + + @property + def held(self) -> bool: + return self.outcome == "renewed" + + @property + def lost(self) -> bool: + return self.outcome == "lost" + + +class AnalysisControlPlane(Protocol): + """Queue ownership for durable analysis jobs. + + Implementations must make every method safe against concurrent workers + without holding a lock across a call: each is a single compare-and-swap or + a read, so a worker process can crash between any two calls and leave only + an expired lease behind. + """ + + def next_eligible_job_id(self, session: Session, *, now: datetime | None = None) -> str | None: + """The oldest job the queue would hand out next, without claiming it.""" + ... + + def claim(self, session: Session, *, worker_id: str) -> JobLease | None: + """Take exclusive ownership of the oldest eligible queued job.""" + ... + + def renew(self, session: Session, lease: JobLease) -> LeaseRenewal: + """Extend ``lease`` and report any cancellation request.""" + ... + + def expired_job_ids(self, session: Session, *, now: datetime | None = None) -> tuple[str, ...]: + """Ids of running jobs whose lease has lapsed, oldest lapse first.""" + ... + + def reclaim(self, session: Session, job: AnalysisJob, *, worker_id: str) -> JobLease | None: + """Take ownership of one expired job, or ``None`` if it moved on.""" + ... + + def update_owned( + self, + session: Session, + *, + job_id: str, + worker_id: str, + values: dict[str, object], + require_cancel_not_requested: bool = False, + ) -> bool: + """Apply ``values`` only while ``worker_id`` still owns the running job.""" + ... + + def cancel_requested(self, session: Session, job_id: str) -> bool: + """Read the durable cancellation flag for ``job_id``.""" + ... + + +class DatabaseAnalysisControlPlane: + """The v1 control plane: the durable ``analysis_jobs`` table itself.""" + + def __init__( + self, + *, + lease_seconds: int, + clock: Callable[[], datetime] = lambda: datetime.now(UTC), + ) -> None: + self.lease_seconds = lease_seconds + self._clock = clock + + def _lease_until(self, now: datetime) -> datetime: + return now + timedelta(seconds=self.lease_seconds) + + @staticmethod + def _execute_dml(session: Session, statement: Executable) -> CursorResult[Any]: + """Execute a guarded UPDATE and expose its ``rowcount``. + + ``Session.execute`` is declared as returning ``Result``, which carries + no ``rowcount``; DML always produces a ``CursorResult``, and every + compare-and-swap here decides ownership from exactly that value. + """ + + return cast(CursorResult[Any], session.execute(statement)) + + # -- claim --------------------------------------------------------------- + + def next_eligible_job_id(self, session: Session, *, now: datetime | None = None) -> str | None: + """The oldest job the queue would hand out next, or ``None``. + + Deliberately separate from :meth:`claim`: this read is the half of a + claim that carries no ownership at all, and naming it makes the race + window between reading a candidate and winning it explicit -- both to a + reader and to a test that needs to hold a candidate across another + worker's claim. + """ + + moment = now if now is not None else self._clock() + return session.scalar( + select(AnalysisJob.id) + .where( + AnalysisJob.status == "queued", + or_(AnalysisJob.next_attempt_at.is_(None), AnalysisJob.next_attempt_at <= moment), + ) + .order_by(AnalysisJob.created_at) + .limit(1) + ) + + def claim(self, session: Session, *, worker_id: str) -> JobLease | None: + """Atomically claim the oldest eligible queued job, or return ``None``. + + A portable compare-and-swap rather than ``SELECT ... FOR UPDATE SKIP + LOCKED``: pick the oldest eligible id, then ``UPDATE ... WHERE id = :id + AND status='queued'``. The ``status='queued'`` predicate is the atomic + guard -- two workers racing for the same row see exactly one non-zero + ``rowcount``; the loser gets ``None`` and polls again. Eligibility also + honours ``next_attempt_at``, so a job serving retry backoff is invisible + to the queue until its delay elapses. + """ + + now = self._clock() + candidate_id = self.next_eligible_job_id(session, now=now) + if candidate_id is None: + return None + expires_at = self._lease_until(now) + result = self._execute_dml( + session, + update(AnalysisJob) + .where(AnalysisJob.id == candidate_id, AnalysisJob.status == "queued") + .values( + status="running", + worker_id=worker_id, + lease_expires_at=expires_at, + started_at=func.coalesce(AnalysisJob.started_at, now), + attempt=AnalysisJob.attempt + 1, + next_attempt_at=None, + updated_at=now, + ), + ) + session.commit() + if result.rowcount == 0: + # Another worker won the compare-and-swap for this row. + return None + claimed = session.get(AnalysisJob, candidate_id) + if claimed is None: + return None + return JobLease( + job_id=candidate_id, + worker_id=worker_id, + expires_at=expires_at, + attempt=claimed.attempt, + ) + + # -- renew --------------------------------------------------------------- + + def renew(self, session: Session, lease: JobLease) -> LeaseRenewal: + """Atomically extend ownership and read back the cancellation flag. + + The guard is ``status='running' AND worker_id = ``: a worker that + has been reclaimed, or whose job reached a terminal state, matches no + row and is told its lease is ``lost`` instead of writing to a job it no + longer owns. Renewal and cancellation observation are one statement, so + a cancellation request accepted between the two can never be missed. + """ + + sqlite_connection: Any = None + sqlite_busy_timeout: int | None = None + + def _restore_sqlite_timeout() -> None: + nonlocal sqlite_connection + connection = sqlite_connection + if connection is None or sqlite_busy_timeout is None: + return + try: + cursor = connection.cursor() + cursor.execute(f"PRAGMA busy_timeout = {sqlite_busy_timeout}") + cursor.close() + except Exception: # noqa: BLE001 - discard a modified connection + session.invalidate() + finally: + sqlite_connection = None + + try: + if session.bind is not None and session.bind.dialect.name == "sqlite": + driver_connection = session.connection().connection.driver_connection + # A pooled connection with no live DBAPI connection has no + # busy timeout to tune; the renewal is still correct, it just + # waits the configured default like any other statement. + if driver_connection is not None: + sqlite_connection = driver_connection + cursor = driver_connection.cursor() + sqlite_busy_timeout = int(cursor.execute("PRAGMA busy_timeout").fetchone()[0]) + # SQLite serializes all writers. If the stage already owns + # the database write lock, a sweeper cannot reclaim the job + # either, so the renewal must not block stage cleanup + # behind that lock. + cursor.execute("PRAGMA busy_timeout = 0") + cursor.close() + now = self._clock() + expires_at = self._lease_until(now) + row = session.execute( + update(AnalysisJob) + .where( + AnalysisJob.id == lease.job_id, + AnalysisJob.status == "running", + AnalysisJob.worker_id == lease.worker_id, + ) + .values(lease_expires_at=expires_at, updated_at=now) + .returning(AnalysisJob.cancel_requested) + .execution_options(synchronize_session=False) + ).first() + if row is None: + _restore_sqlite_timeout() + session.rollback() + return LeaseRenewal(outcome="lost") + _restore_sqlite_timeout() + session.commit() + return LeaseRenewal(outcome="renewed", cancel_requested=bool(row[0]), expires_at=expires_at) + except Exception as exc: # noqa: BLE001 - classified here, re-raised to the caller + _restore_sqlite_timeout() + session.rollback() + # A SQLite writer prevents every other SQLite writer, including a + # stale sweeper. Skipping that transient pulse avoids a false retry; + # PostgreSQL renewals remain fully independent and guarded. + if session.bind is not None and session.bind.dialect.name == "sqlite" and "locked" in str(exc).lower(): + logger.debug("SQLite analysis lease renewal skipped while the stage held the write lock") + return LeaseRenewal(outcome="deferred") + raise + finally: + _restore_sqlite_timeout() + + # -- expiry and reclaim -------------------------------------------------- + + def expired_job_ids(self, session: Session, *, now: datetime | None = None) -> tuple[str, ...]: + """Ids of running jobs whose lease has lapsed, oldest lapse first. + + Read-only, and not itself a claim: a caller must still win + :meth:`reclaim` for each id before acting on it, because another sweeper + may reconcile the same row between this scan and that call. + """ + + moment = now if now is not None else self._clock() + return tuple( + session.scalars( + select(AnalysisJob.id) + .where( + AnalysisJob.status == "running", + AnalysisJob.lease_expires_at.is_not(None), + AnalysisJob.lease_expires_at < moment, + ) + .order_by(AnalysisJob.lease_expires_at, AnalysisJob.created_at) + ) + ) + + def reclaim(self, session: Session, job: AnalysisJob, *, worker_id: str) -> JobLease | None: + """Take ownership of one expired job, or ``None`` if it moved on. + + The guard pins the *exact* prior owner and lease instant as well as + requiring the lease to still be lapsed, so an active lease can never be + stolen: a renewal landing between the scan and this statement moves + ``lease_expires_at`` and the compare-and-swap matches nothing. + """ + + now = self._clock() + expires_at = self._lease_until(now) + result = self._execute_dml( + session, + update(AnalysisJob) + .where( + AnalysisJob.id == job.id, + AnalysisJob.status == "running", + AnalysisJob.worker_id == job.worker_id, + AnalysisJob.lease_expires_at == job.lease_expires_at, + AnalysisJob.lease_expires_at < now, + ) + .values(worker_id=worker_id, lease_expires_at=expires_at, updated_at=now) + .execution_options(synchronize_session="fetch"), + ) + if result.rowcount == 0: + session.rollback() + return None + return JobLease(job_id=job.id, worker_id=worker_id, expires_at=expires_at, attempt=job.attempt) + + # -- ownership-guarded mutation ------------------------------------------ + + def update_owned( + self, + session: Session, + *, + job_id: str, + worker_id: str, + values: dict[str, object], + require_cancel_not_requested: bool = False, + ) -> bool: + """Apply ``values`` only while ``worker_id`` owns the running job. + + Returns ``True`` when the guard matched and ``False`` when ownership was + lost. Transactional recovery from a lost guard belongs to the caller: + this method neither commits nor rolls back, so the caller can decide + whether a miss means abandon, cancel, or reconcile. + """ + + ownership = [ + AnalysisJob.id == job_id, + AnalysisJob.worker_id == worker_id, + AnalysisJob.status == "running", + ] + if require_cancel_not_requested: + ownership.append(AnalysisJob.cancel_requested.is_(False)) + with session.no_autoflush: + result = self._execute_dml( + session, + update(AnalysisJob).where(*ownership).values(**values).execution_options(synchronize_session="fetch"), + ) + return bool(result.rowcount) + + # -- cancellation -------------------------------------------------------- + + def cancel_requested(self, session: Session, job_id: str) -> bool: + """Read the durable cancellation flag for ``job_id``. + + Cancellation is observed from the row rather than carried on the lease, + so a request accepted by the API after this worker claimed the job is + still seen at the next cooperative check. + """ + + return bool(session.scalar(select(AnalysisJob.cancel_requested).where(AnalysisJob.id == job_id))) diff --git a/apps/backend/tests/test_analysis_worker.py b/apps/backend/tests/test_analysis_worker.py index d186941b..9325c6f5 100644 --- a/apps/backend/tests/test_analysis_worker.py +++ b/apps/backend/tests/test_analysis_worker.py @@ -544,9 +544,8 @@ def test_sqlite_write_lock_does_not_leave_a_heartbeat_thread_running(session_fac heartbeat_interval_seconds=0.02, ) with session_factory() as claim_session: - job = worker._claim(claim_session) - assert job is not None - job_id = job.id + lease = worker.control_plane.claim(claim_session, worker_id="worker-a") + assert lease is not None blocker = session_factory() try: @@ -554,7 +553,7 @@ def test_sqlite_write_lock_does_not_leave_a_heartbeat_thread_running(session_fac update(RepositoryRecord).where(RepositoryRecord.id == record_id).values(updated_at=datetime.now(UTC)) ) state = _HeartbeatState() - thread = threading.Thread(target=worker._heartbeat_once, args=(job_id, state)) + thread = threading.Thread(target=worker._heartbeat_once, args=(lease, state)) thread.start() thread.join(timeout=1) From c113b6c361bd3fb93bb5893e7164469e1ce26150 Mon Sep 17 00:00:00 2001 From: Shaurya K Sharma Date: Sat, 29 Aug 2026 22:28:43 +0100 Subject: [PATCH 2/4] refactor(scale): move the analysis worker loop out of the API app app.main did not merely host a worker, it *was* the control loop: it minted the ownership token, built the worker, owned the poll cadence, decided when to sweep stale leases, and joined the thread on shutdown. Nothing outside a FastAPI lifespan could run a worker without copying that policy, which is the tight coupling #324 exists to remove. AnalysisWorkerRunner now owns it. run_forever is a plain blocking call, so the loop a standalone worker process needs already exists here rather than being locked inside an async context manager; threading is an implementation detail of this runner, not of the boundary. app.main is reduced to start/stop, and the in-process single-worker path is preserved exactly as #324 requires during migration. Building the standalone deployment stays with #210. new_worker_id moves alongside it, since worker identity is what every control-plane ownership guard compares; its test moves with it and now asserts uniqueness across many tokens rather than two. Refs #324, #210 --- apps/backend/app/core/config.py | 13 +- apps/backend/app/main.py | 79 +++--------- apps/backend/app/workers/runner.py | 197 +++++++++++++++++++++++++++++ apps/backend/tests/test_system.py | 15 ++- 4 files changed, 234 insertions(+), 70 deletions(-) create mode 100644 apps/backend/app/workers/runner.py diff --git a/apps/backend/app/core/config.py b/apps/backend/app/core/config.py index 2773d7fa..6e4e6dab 100644 --- a/apps/backend/app/core/config.py +++ b/apps/backend/app/core/config.py @@ -74,12 +74,13 @@ class Settings(BaseSettings): rate_limit_auth_per_minute: int = 10 rate_limit_ai_per_minute: int = 20 rate_limit_heavy_per_minute: int = 30 - # Durable analysis-job worker (#93). ``analysis_worker_autostart`` gates the - # background daemon thread started in ``app.main``'s lifespan; tests set it - # False so they drive ``AnalysisWorker.run_once()`` deterministically instead - # of racing a real thread. The poll interval bounds how long the loop sleeps - # between empty polls; the lease bounds how long a claimed job is owned before - # a future stale-job sweep may reclaim it. + # Durable analysis-job worker (#93, #324). ``analysis_worker_autostart`` + # gates the in-process ``AnalysisWorkerRunner`` started from ``app.main``'s + # lifespan; tests set it False so they drive ``AnalysisWorker.run_once()`` + # deterministically instead of racing a real thread. The poll interval bounds + # how long the runner loop sleeps between empty polls; the lease bounds how + # long a claimed job is owned before the control plane lets a stale-job sweep + # reclaim it. analysis_worker_autostart: bool = True analysis_job_poll_interval_seconds: int = 5 analysis_job_lease_seconds: int = 300 diff --git a/apps/backend/app/main.py b/apps/backend/app/main.py index 6059b23f..9efeebc6 100644 --- a/apps/backend/app/main.py +++ b/apps/backend/app/main.py @@ -3,10 +3,8 @@ from os import getpid from pathlib import Path import logging -import threading from time import perf_counter -from typing import Any, Literal -from uuid import uuid4 +from typing import TYPE_CHECKING, Any, Literal from fastapi import Depends, FastAPI, Request, status from fastapi.middleware.cors import CORSMiddleware @@ -33,9 +31,10 @@ from app.core.security_headers import SecurityHeadersMiddleware from app.models.base import Base -logger = logging.getLogger(__name__) +if TYPE_CHECKING: + from app.workers.runner import AnalysisWorkerRunner -_ANALYSIS_STALE_SWEEP_INTERVAL = 10 +logger = logging.getLogger(__name__) _READINESS_SCHEMA = { "type": "object", @@ -94,63 +93,28 @@ def check_storage_ready() -> bool: return True -def _analysis_worker_id() -> str: - """Return a process-observable, globally unique worker ownership token.""" - - return f"analysis-worker-{getpid()}-{uuid4().hex}" - +def _start_analysis_worker() -> "AnalysisWorkerRunner | None": + """Start the in-process analysis worker, unless it is switched off (#324). -def _start_analysis_worker() -> tuple[threading.Thread, threading.Event, Any] | None: - """Start the durable analysis worker on a daemon thread (#93). + The API process hosts a worker for compatibility; it does not *own* the + queue. Worker identity, poll cadence, stale-sweep cadence and shutdown all + live in ``app.workers.runner`` behind the control-plane boundary, so a + standalone worker process would reuse the identical loop rather than + reimplementing this function. - The loop claims and runs one queued job per iteration, sleeping only when the - queue is empty so a backlog drains promptly. It is gated by - ``analysis_worker_autostart`` so tests drive ``run_once`` deterministically - instead of racing this thread. + ``analysis_worker_autostart`` gates the runner so tests drive + ``AnalysisWorker.run_once`` deterministically instead of racing a thread. """ settings = get_settings() if not settings.analysis_worker_autostart: return None - from app.core.database import SessionLocal - from app.workers.analysis_worker import AnalysisWorker - - worker = AnalysisWorker( - SessionLocal, - worker_id=_analysis_worker_id(), - lease_seconds=settings.analysis_job_lease_seconds, - max_repository_source_bytes=settings.analysis_max_repository_source_bytes, - max_process_rss_bytes=settings.analysis_max_process_rss_bytes, - max_analysis_seconds=settings.analysis_max_duration_seconds, - ) - stop_event = threading.Event() - - def _loop() -> None: - polls_since_sweep = 0 - while not stop_event.is_set(): - try: - claimed = worker.run_once() - polls_since_sweep += 1 - if polls_since_sweep >= _ANALYSIS_STALE_SWEEP_INTERVAL: - worker.sweep_stale() - polls_since_sweep = 0 - except Exception: # noqa: BLE001 - a single bad job must not kill the loop - logger.exception("Analysis worker iteration failed") - claimed = False - if not claimed: - stop_event.wait(settings.analysis_job_poll_interval_seconds) - - # Reclaim jobs orphaned by a previous hard process exit immediately on - # startup; the loop repeats the sweep periodically for later crashes. - try: - worker.sweep_stale() - except Exception: # noqa: BLE001 - stale cleanup must not prevent API startup - logger.exception("Initial stale analysis-job sweep failed") + from app.workers.runner import build_analysis_worker_runner - thread = threading.Thread(target=_loop, name="analysis-worker", daemon=True) - thread.start() - return thread, stop_event, worker + runner = build_analysis_worker_runner(settings) + runner.start() + return runner @asynccontextmanager @@ -174,15 +138,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: Base.metadata.create_all(bind=database.engine) else: ensure_schema_in_sync(database.engine, app_env=settings.app_env) - worker_handle = _start_analysis_worker() + analysis_worker_runner = _start_analysis_worker() try: yield finally: - if worker_handle is not None: - thread, stop_event, worker = worker_handle - stop_event.set() - worker.shutdown() - thread.join(timeout=10) + if analysis_worker_runner is not None: + analysis_worker_runner.stop() aclose = getattr(app.state.rate_limit_store, "aclose", None) if aclose is not None: await aclose() diff --git a/apps/backend/app/workers/runner.py b/apps/backend/app/workers/runner.py new file mode 100644 index 00000000..b7df402a --- /dev/null +++ b/apps/backend/app/workers/runner.py @@ -0,0 +1,197 @@ +"""The in-process compatibility runner for the analysis control plane (#324). + +#324 requires the current single-worker path to stay available *behind* the new +boundary during migration. This module is that path. + +Before this module, ``app.main`` constructed the worker, minted its ownership +token, owned the poll loop, decided the stale-sweep cadence, and joined the +thread on shutdown -- so the API process did not merely *host* a worker, it +*was* the control loop. Nothing outside a FastAPI lifespan could run a worker +without copying that policy. + +``AnalysisWorkerRunner`` owns that policy instead. ``app.main`` now only starts +and stops it, and :meth:`AnalysisWorkerRunner.run_forever` is a plain blocking +call, so the same loop a future standalone worker process needs is already +here -- a ``__main__`` that builds a runner and calls ``run_forever`` adds no +new queue policy. Building that deployment is #210's remaining work and is +deliberately not done here. + +Threading is an implementation detail of *this* runner, not of the boundary: the +claim/lease contract in ``app.workers.control_plane`` is process-agnostic, and a +separate process would use the identical contract without a daemon thread. +""" + +from __future__ import annotations + +import logging +import threading +from collections.abc import Callable +from os import getpid +from uuid import uuid4 + +from sqlalchemy.orm import Session + +from app.core.config import Settings, get_settings +from app.workers.analysis_worker import AnalysisWorker + +logger = logging.getLogger(__name__) + +#: Empty-queue polls between stale-lease sweeps. Reconciling an expired lease is +#: a scan plus a compare-and-swap per stale row, so it is far cheaper than an +#: analysis but not free; once per N polls keeps recovery prompt without turning +#: an idle worker into a busy sweeper. +DEFAULT_STALE_SWEEP_INTERVAL_POLLS = 10 + +#: How long ``stop`` waits for the loop thread to leave its current iteration. +DEFAULT_SHUTDOWN_TIMEOUT_SECONDS = 10.0 + + +def new_worker_id(pid: int | None = None) -> str: + """Mint a process-observable, globally unique worker ownership token. + + The pid makes an owner traceable to a process in logs; the uuid makes two + workers in the *same* process (or in two containers that happen to share a + pid namespace) distinct owners. Uniqueness is what the control plane's + ownership guards rest on, so it must not depend on the pid alone. The result + fits ``analysis_jobs.worker_id`` (64 characters). + """ + + return f"analysis-worker-{pid if pid is not None else getpid()}-{uuid4().hex}" + + +class AnalysisWorkerRunner: + """Drive one :class:`AnalysisWorker` against the queue until stopped. + + The loop claims and runs one job per iteration and sleeps *only* when the + queue is empty, so a backlog drains without an artificial poll delay between + jobs. A failed iteration is logged and the loop continues: one bad job must + never stop the worker, and the job's own bounded-retry and stale-lease paths + already decide what happens to it. + """ + + def __init__( + self, + worker: AnalysisWorker, + *, + poll_interval_seconds: float, + stale_sweep_interval_polls: int = DEFAULT_STALE_SWEEP_INTERVAL_POLLS, + ) -> None: + self.worker = worker + self.poll_interval_seconds = poll_interval_seconds + self.stale_sweep_interval_polls = stale_sweep_interval_polls + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + @property + def worker_id(self) -> str: + return self.worker.worker_id + + def sweep_on_start(self) -> None: + """Reclaim jobs orphaned by a previous hard process exit. + + A crash leaves a running row with a lease nobody will renew. Sweeping + once at startup recovers those immediately instead of waiting out the + first periodic sweep. A failure here must not prevent the process from + starting, so it is logged rather than raised. + """ + + try: + self.worker.sweep_stale() + except Exception: # noqa: BLE001 - stale cleanup must not block startup + logger.exception("Initial stale analysis-job sweep failed") + + def run_forever(self) -> None: + """Run the claim/sweep loop on the calling thread until :meth:`stop`. + + This is the whole control loop. :meth:`start` runs it on a daemon thread + for the in-process path; a standalone worker process would call it + directly. + """ + + polls_since_sweep = 0 + while not self._stop.is_set(): + try: + claimed = self.worker.run_once() + polls_since_sweep += 1 + if polls_since_sweep >= self.stale_sweep_interval_polls: + self.worker.sweep_stale() + polls_since_sweep = 0 + except Exception: # noqa: BLE001 - a single bad job must not kill the loop + logger.exception("Analysis worker iteration failed") + claimed = False + if not claimed: + self._stop.wait(self.poll_interval_seconds) + + def start(self) -> None: + """Sweep once, then run the loop on a daemon thread.""" + + if self._thread is not None: + raise RuntimeError("analysis worker runner is already started") + self.sweep_on_start() + self._stop.clear() + self._thread = threading.Thread(target=self.run_forever, name="analysis-worker", daemon=True) + self._thread.start() + + def stop(self, timeout: float = DEFAULT_SHUTDOWN_TIMEOUT_SECONDS) -> None: + """Signal the loop and the worker's heartbeats, then join the thread. + + Both signals are needed and ordered: ``_stop`` ends the loop after the + current iteration, and ``worker.shutdown()`` releases a stage heartbeat + that would otherwise keep renewing a lease while the process exits. A + thread still alive after ``timeout`` is a daemon, so it cannot block + interpreter exit; it is reported rather than killed. + """ + + self._stop.set() + self.worker.shutdown() + thread = self._thread + self._thread = None + if thread is None: + return + thread.join(timeout=timeout) + if thread.is_alive(): + logger.warning( + "Analysis worker loop did not stop within the shutdown timeout", + extra={"worker_id": self.worker_id, "timeout_seconds": timeout}, + ) + + +def build_analysis_worker( + settings: Settings, + session_factory: Callable[[], Session], + *, + worker_id: str | None = None, +) -> AnalysisWorker: + """Construct the configured executor for the durable analysis queue.""" + + return AnalysisWorker( + session_factory, + worker_id=worker_id or new_worker_id(), + lease_seconds=settings.analysis_job_lease_seconds, + max_repository_source_bytes=settings.analysis_max_repository_source_bytes, + max_process_rss_bytes=settings.analysis_max_process_rss_bytes, + max_analysis_seconds=settings.analysis_max_duration_seconds, + ) + + +def build_analysis_worker_runner( + settings: Settings | None = None, + session_factory: Callable[[], Session] | None = None, +) -> AnalysisWorkerRunner: + """Assemble the configured in-process runner. + + Imported lazily by callers that only need it when the worker actually + autostarts, which keeps the database session factory out of import order for + processes that never run a worker. + """ + + resolved_settings = settings if settings is not None else get_settings() + if session_factory is None: + from app.core.database import SessionLocal + + session_factory = SessionLocal + worker = build_analysis_worker(resolved_settings, session_factory) + return AnalysisWorkerRunner( + worker, + poll_interval_seconds=resolved_settings.analysis_job_poll_interval_seconds, + ) diff --git a/apps/backend/tests/test_system.py b/apps/backend/tests/test_system.py index c36121ce..9a893d30 100644 --- a/apps/backend/tests/test_system.py +++ b/apps/backend/tests/test_system.py @@ -178,13 +178,18 @@ def test_settings_rejects_invalid_log_format(): Settings(log_format="pretty") -def test_production_analysis_worker_ids_are_unique_with_the_same_pid(monkeypatch): - import app.main as main_module +def test_production_analysis_worker_ids_are_unique_with_the_same_pid(): + """Two workers in one process must be two distinct queue owners (#324). + + Every control-plane ownership guard is ``worker_id`` equality, so a token + that collided between two workers in the same process would let each mutate + the other's job. + """ - monkeypatch.setattr(main_module, "getpid", lambda: 42) + from app.workers.runner import new_worker_id - first = main_module._analysis_worker_id() - second = main_module._analysis_worker_id() + first = new_worker_id(pid=42) + second = new_worker_id(pid=42) assert first != second assert first.startswith("analysis-worker-42-") From d3b07de225bc1868c0a71fcb972ca0fe266f98b9 Mon Sep 17 00:00:00 2001 From: Shaurya K Sharma Date: Sat, 29 Aug 2026 22:28:58 +0100 Subject: [PATCH 3/4] test(scale): cover queue ownership, leases and cancellation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 31 cases over the boundary itself, not the extraction pipeline: claiming and eligibility (including that next_attempt_at backoff hides a job from the queue), the duplicate-claim race, separate-job claiming, expired-lease reclaim, active- lease protection, ownership enforcement on both mutation and renewal, the lost-lease handoff signal, cancellation visibility across a reclaim, the cancel-not-requested guard that keeps cancellation idempotent, and the runner's drain/sweep/failure-isolation/shutdown behaviour. Every race is expressed as an explicit interleaving or a threading.Barrier; nothing sleeps waiting for a race, so a slow machine cannot turn a correctness assertion into a flake. The duplicate-claim case pins the candidate each worker read so both genuinely reach the compare-and-swap — verified by mutation: deleting the status='queued' guard fails it, and an earlier version of the test that let the loser re-read the queue did not. next_eligible_job_id is split out of claim for that reason: the candidate read is the half of a claim that carries no ownership, and naming it makes the race window visible to a reader as well as reachable by a test. The threaded race is gated on PARTHA_TEST_PG_URL, following the repository's established pattern — SQLite serialises writers, so only a real MVCC server exercises two claims genuinely in flight. Refs #324, #210 --- .../tests/test_analysis_control_plane.py | 842 ++++++++++++++++++ 1 file changed, 842 insertions(+) create mode 100644 apps/backend/tests/test_analysis_control_plane.py diff --git a/apps/backend/tests/test_analysis_control_plane.py b/apps/backend/tests/test_analysis_control_plane.py new file mode 100644 index 00000000..6c8ab796 --- /dev/null +++ b/apps/backend/tests/test_analysis_control_plane.py @@ -0,0 +1,842 @@ +"""The analysis queue and control-plane boundary (#324). + +These tests exercise ``app.workers.control_plane`` and ``app.workers.runner`` +directly, without running the extraction pipeline: the point is *who owns a +job*, not what analysing it produces. Pipeline behaviour is already covered by +``test_analysis_worker.py``, and this file must not become a second copy of it. + +Determinism: every race here is expressed as an explicit interleaving (both +sides read, then both sides write) or synchronised with a ``threading.Barrier``. +Nothing sleeps waiting for a race to happen, so a slow machine cannot turn a +correctness assertion into a flake. The one genuinely threaded race is gated on +a real PostgreSQL server, matching the repository's established pattern for +concurrency that SQLite's single-writer lock cannot represent. +""" + +from __future__ import annotations + +import os +import threading +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, func, select +from sqlalchemy.orm import sessionmaker + +from app.core.database import register_sqlite_foreign_key_enforcement +from app.models import RepositoryRecord, User +from app.models.analysis_job import AnalysisJob +from app.models.base import Base +from app.services.analysis_job_service import ANALYSIS_CONFIG_HASH +from app.workers.control_plane import ( + DatabaseAnalysisControlPlane, + JobLease, + lease_expired, +) +from app.workers.runner import AnalysisWorkerRunner, new_worker_id + +UPLOAD_REVISION = "sha256:" + "c" * 64 +PG_URL = os.environ.get("PARTHA_TEST_PG_URL") + + +@pytest.fixture() +def session_factory(tmp_path): + register_sqlite_foreign_key_enforcement() + engine = create_engine(f"sqlite:///{tmp_path / 'control-plane.db'}") + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine, expire_on_commit=False) + yield factory + engine.dispose() + + +def _owner(session) -> User: + owner = User(id=str(uuid4()), email=f"{uuid4().hex}@example.com", password_hash=None) + session.add(owner) + session.commit() + return owner + + +def _repository(session, owner: User) -> RepositoryRecord: + record = RepositoryRecord( + id=str(uuid4()), + owner_id=owner.id, + name="repo", + source="upload", + revision_kind="upload", + revision_value=UPLOAD_REVISION, + local_path="/x", + status="analysing", + file_tree=[], + ) + session.add(record) + session.commit() + return record + + +def _queued_job(session, record: RepositoryRecord, **overrides) -> AnalysisJob: + """Insert a queued job row directly -- the queue's input, not the pipeline's.""" + + values: dict[str, object] = { + "id": str(uuid4()), + "repository_id": record.id, + "owner_id": record.owner_id, + "revision_kind": record.revision_kind, + "revision_value": record.revision_value, + "config_hash": ANALYSIS_CONFIG_HASH, + "status": "queued", + "attempt": 0, + } + values.update(overrides) + job = AnalysisJob(**values) + session.add(job) + session.commit() + return job + + +def _bootstrap(factory, **overrides) -> tuple[str, str]: + """Create one owner, repository and queued job; return (record_id, job_id).""" + + with factory() as session: + owner = _owner(session) + record = _repository(session, owner) + job = _queued_job(session, record, **overrides) + return record.id, job.id + + +def _plane(lease_seconds: int = 60, clock=None) -> DatabaseAnalysisControlPlane: + if clock is None: + return DatabaseAnalysisControlPlane(lease_seconds=lease_seconds) + return DatabaseAnalysisControlPlane(lease_seconds=lease_seconds, clock=clock) + + +# -- AC1: jobs are claimed through an explicit lease path -------------------- + + +def test_claim_returns_a_lease_and_marks_the_job_running(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + lease = plane.claim(session, worker_id="worker-a") + + assert lease is not None + assert lease.job_id == job_id + assert lease.worker_id == "worker-a" + assert lease.attempt == 1 + + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.status == "running" + assert job.worker_id == "worker-a" + assert job.lease_expires_at is not None + assert job.started_at is not None + assert job.next_attempt_at is None + + +def test_claim_returns_none_when_the_queue_is_empty(session_factory): + with session_factory() as session: + owner = _owner(session) + _repository(session, owner) + + with session_factory() as session: + assert _plane().claim(session, worker_id="worker-a") is None + + +def test_claim_skips_a_job_still_serving_retry_backoff(session_factory): + """``next_attempt_at`` is queue eligibility, not just a record of intent.""" + + future = datetime.now(UTC) + timedelta(seconds=300) + _, job_id = _bootstrap(session_factory, next_attempt_at=future) + plane = _plane() + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is None + + # Once the delay elapses the same job becomes claimable, unchanged. + later = _plane(clock=lambda: datetime.now(UTC) + timedelta(seconds=600)) + with session_factory() as session: + lease = later.claim(session, worker_id="worker-a") + assert lease is not None and lease.job_id == job_id + + +def test_claim_takes_the_oldest_eligible_job_first(session_factory): + with session_factory() as session: + owner = _owner(session) + record = _repository(session, owner) + older = _queued_job(session, record, created_at=datetime.now(UTC) - timedelta(minutes=5)) + # A second repository, because one repository may hold only one + # effective-identity job at a time. + other_record = _repository(session, owner) + _queued_job(session, other_record, created_at=datetime.now(UTC)) + older_id = older.id + + with session_factory() as session: + lease = _plane().claim(session, worker_id="worker-a") + + assert lease is not None and lease.job_id == older_id + + +# -- AC2: duplicate claims and expired leases -------------------------------- + + +class _PinnedCandidatePlane(DatabaseAnalysisControlPlane): + """A control plane frozen just after it read its candidate. + + A claim is a candidate read followed by a compare-and-swap. The loser of a + real race is a worker whose read happened *before* the winner committed, so + reproducing the race means holding that stale candidate across the winner's + claim. Pinning the id does exactly that and changes nothing else: the swap + under test is the unmodified production statement. + """ + + def __init__(self, candidate_id: str, **kwargs) -> None: + super().__init__(**kwargs) + self._candidate_id = candidate_id + + def next_eligible_job_id(self, session, *, now=None): + return self._candidate_id + + +def test_two_workers_racing_one_job_produce_exactly_one_owner(session_factory): + """The claim guard, proved by an explicit interleaving. + + Both workers resolve the same candidate before either swaps, which is + precisely the window a ``SELECT`` then ``UPDATE`` claim has to survive. The + ``status='queued'`` predicate is what makes the loser's write match no row. + """ + + _, job_id = _bootstrap(session_factory) + + session_a = session_factory() + session_b = session_factory() + try: + # Both read the queue while the job is still queued. + candidate_a = _plane().next_eligible_job_id(session_a) + candidate_b = _plane().next_eligible_job_id(session_b) + assert candidate_a == candidate_b == job_id + + # Both now swap, each still holding the candidate it read. + lease_a = _PinnedCandidatePlane(candidate_a, lease_seconds=60).claim(session_a, worker_id="worker-a") + lease_b = _PinnedCandidatePlane(candidate_b, lease_seconds=60).claim(session_b, worker_id="worker-b") + finally: + session_a.close() + session_b.close() + + winners = [lease for lease in (lease_a, lease_b) if lease is not None] + assert len(winners) == 1, "both workers claimed the same job" + + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.status == "running" + assert job.worker_id == winners[0].worker_id + # The loser must not have inflated the attempt budget on its way past. + assert job.attempt == 1 + + +def test_a_second_claim_of_a_running_job_takes_nothing(session_factory): + """The swap guard alone, with no candidate-selection help.""" + + _, job_id = _bootstrap(session_factory) + + with session_factory() as session: + assert _plane().claim(session, worker_id="worker-a") is not None + + # worker-b swaps against a candidate that is no longer queued. + with session_factory() as session: + assert _PinnedCandidatePlane(job_id, lease_seconds=60).claim(session, worker_id="worker-b") is None + + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.worker_id == "worker-a" + assert job.attempt == 1 + + +def test_two_workers_claim_two_separate_jobs(session_factory): + with session_factory() as session: + owner = _owner(session) + first = _queued_job(session, _repository(session, owner)) + second = _queued_job(session, _repository(session, owner)) + job_ids = {first.id, second.id} + + with session_factory() as session: + lease_a = _plane().claim(session, worker_id="worker-a") + with session_factory() as session: + lease_b = _plane().claim(session, worker_id="worker-b") + + assert lease_a is not None and lease_b is not None + assert {lease_a.job_id, lease_b.job_id} == job_ids + + +def test_an_expired_lease_is_reclaimable_by_another_worker(session_factory): + _, job_id = _bootstrap(session_factory) + expired_clock = datetime.now(UTC) - timedelta(hours=1) + stale = _plane(clock=lambda: expired_clock) + + with session_factory() as session: + assert stale.claim(session, worker_id="worker-a") is not None + + fresh = _plane() + with session_factory() as session: + assert fresh.expired_job_ids(session) == (job_id,) + job = session.get(AnalysisJob, job_id) + reclaimed = fresh.reclaim(session, job, worker_id="worker-b") + session.commit() + + assert reclaimed is not None + assert reclaimed.worker_id == "worker-b" + + with session_factory() as reader: + row = reader.get(AnalysisJob, job_id) + assert row.worker_id == "worker-b" + assert row.status == "running" + # Reclaim transfers ownership only; the attempt budget is retry policy + # and must not be spent by a handoff. + assert row.attempt == 1 + + +def test_an_active_lease_is_never_stolen(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane(lease_seconds=3600) + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is not None + + with session_factory() as session: + assert plane.expired_job_ids(session) == () + job = session.get(AnalysisJob, job_id) + assert plane.reclaim(session, job, worker_id="worker-b") is None + + with session_factory() as reader: + assert reader.get(AnalysisJob, job_id).worker_id == "worker-a" + + +def test_a_renewal_between_scan_and_reclaim_defeats_the_reclaim(session_factory): + """The reclaim guard pins the exact lease instant it observed. + + A sweeper that read an expired row must not act on that stale read if the + owner renewed in between -- otherwise a live worker loses its job to a + sweep it had already outrun. + """ + + _, job_id = _bootstrap(session_factory) + expired_clock = datetime.now(UTC) - timedelta(hours=1) + stale = _plane(clock=lambda: expired_clock) + + with session_factory() as session: + lease = stale.claim(session, worker_id="worker-a") + assert lease is not None + + sweeper = _plane() + with session_factory() as scan_session: + # The sweeper observes the row while the lease is still lapsed. + assert sweeper.expired_job_ids(scan_session) == (job_id,) + observed = scan_session.get(AnalysisJob, job_id) + assert observed.worker_id == "worker-a" + + # The rightful owner renews after that scan, from its own session. + with session_factory() as owner_session: + assert _plane().renew(owner_session, lease).held is True + + # The sweeper now acts on what it read. Its guard pins that lease + # instant, which no longer matches the row, so it takes nothing. + assert sweeper.reclaim(scan_session, observed, worker_id="worker-b") is None + + with session_factory() as reader: + assert reader.get(AnalysisJob, job_id).worker_id == "worker-a" + + +def test_lease_expired_compares_naive_and_aware_timestamps(session_factory): + """SQLite hands back naive datetimes; PostgreSQL hands back aware ones.""" + + aware = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) + naive = datetime(2026, 1, 1, 12, 0) + + assert lease_expired(naive, aware + timedelta(seconds=1)) is True + assert lease_expired(naive, aware - timedelta(seconds=1)) is False + assert lease_expired(aware, naive + timedelta(seconds=1)) is True + assert lease_expired(aware, naive - timedelta(seconds=1)) is False + + +# -- ownership enforcement --------------------------------------------------- + + +def test_a_non_owner_cannot_mutate_another_workers_job(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is not None + + with session_factory() as session: + held = plane.update_owned( + session, + job_id=job_id, + worker_id="worker-b", + values={"status": "completed", "progress": 100}, + ) + session.rollback() + + assert held is False + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.status == "running" + assert job.worker_id == "worker-a" + assert job.progress == 0 + + +def test_a_non_owner_cannot_renew_another_workers_lease(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + owned = plane.claim(session, worker_id="worker-a") + assert owned is not None + + impostor = JobLease(job_id=job_id, worker_id="worker-b", expires_at=owned.expires_at, attempt=1) + with session_factory() as session: + renewal = plane.renew(session, impostor) + + assert renewal.lost is True + with session_factory() as reader: + assert reader.get(AnalysisJob, job_id).worker_id == "worker-a" + + +def test_a_worker_that_lost_its_lease_is_told_so_on_the_next_renewal(session_factory): + """The handoff signal: renewal is how a displaced owner finds out.""" + + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + lease = plane.claim(session, worker_id="worker-a") + assert lease is not None + + with session_factory() as session: + assert plane.renew(session, lease).held is True + + # A reclaim hands the job to worker-b. + with session_factory() as session: + job = session.get(AnalysisJob, job_id) + job.lease_expires_at = datetime.now(UTC) - timedelta(hours=1) + session.commit() + assert plane.reclaim(session, job, worker_id="worker-b") is not None + session.commit() + + with session_factory() as session: + assert plane.renew(session, lease).lost is True + + +def test_a_terminal_job_cannot_be_renewed_or_mutated(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + lease = plane.claim(session, worker_id="worker-a") + assert lease is not None + + with session_factory() as session: + job = session.get(AnalysisJob, job_id) + job.status = "completed" + session.commit() + + with session_factory() as session: + assert plane.renew(session, lease).lost is True + assert ( + plane.update_owned(session, job_id=job_id, values={"progress": 50}, worker_id="worker-a") is False + ) + session.rollback() + + with session_factory() as reader: + assert reader.get(AnalysisJob, job_id).status == "completed" + + +# -- AC3: cancellation across the boundary ----------------------------------- + + +def test_a_renewal_reports_a_cancellation_request(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + lease = plane.claim(session, worker_id="worker-a") + assert lease is not None + + with session_factory() as session: + assert plane.renew(session, lease).cancel_requested is False + assert plane.cancel_requested(session, job_id) is False + + with session_factory() as session: + session.get(AnalysisJob, job_id).cancel_requested = True + session.commit() + + with session_factory() as session: + renewal = plane.renew(session, lease) + assert renewal.held is True + assert renewal.cancel_requested is True + assert plane.cancel_requested(session, job_id) is True + + +def test_a_cancellation_request_survives_a_reclaim_handoff(session_factory): + """Cancellation must not be dropped when ownership moves.""" + + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is not None + + with session_factory() as session: + job = session.get(AnalysisJob, job_id) + job.cancel_requested = True + job.lease_expires_at = datetime.now(UTC) - timedelta(hours=1) + session.commit() + + reclaimed = plane.reclaim(session, job, worker_id="worker-b") + session.commit() + assert reclaimed is not None + + with session_factory() as session: + assert plane.cancel_requested(session, job_id) is True + assert plane.renew(session, reclaimed).cancel_requested is True + + +def test_cancellation_is_not_resurrected_after_a_cancelled_job_is_reclaimed(session_factory): + """A cancelled job is terminal: no later reclaim can put it back to work.""" + + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is not None + + with session_factory() as session: + job = session.get(AnalysisJob, job_id) + job.status = "cancelled" + job.worker_id = None + job.cancel_requested = False + job.lease_expires_at = None + job.completed_at = datetime.now(UTC) + session.commit() + + with session_factory() as session: + # It is neither claimable nor sweepable, so no worker can own it again. + assert plane.claim(session, worker_id="worker-b") is None + assert plane.expired_job_ids(session) == () + + with session_factory() as reader: + assert reader.get(AnalysisJob, job_id).status == "cancelled" + + +def test_the_cancel_not_requested_guard_refuses_to_complete_a_cancelling_job(session_factory): + """``require_cancel_not_requested`` is what makes cancellation idempotent. + + A completion that raced an accepted cancellation must lose, so the request + cannot be silently discarded by a worker finishing a moment later. + """ + + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + assert plane.claim(session, worker_id="worker-a") is not None + + with session_factory() as session: + session.get(AnalysisJob, job_id).cancel_requested = True + session.commit() + + with session_factory() as session: + guarded = plane.update_owned( + session, + job_id=job_id, + worker_id="worker-a", + values={"status": "completed"}, + require_cancel_not_requested=True, + ) + session.rollback() + # The same worker may still act on the job -- it only may not pretend + # the cancellation never happened. + unguarded = plane.update_owned( + session, + job_id=job_id, + worker_id="worker-a", + values={"status": "cancelled", "cancel_requested": False}, + ) + session.commit() + + assert guarded is False + assert unguarded is True + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.status == "cancelled" + assert job.cancel_requested is False + + +def test_repeated_cancellation_reads_are_idempotent(session_factory): + _, job_id = _bootstrap(session_factory) + plane = _plane() + + with session_factory() as session: + lease = plane.claim(session, worker_id="worker-a") + session.get(AnalysisJob, job_id).cancel_requested = True + session.commit() + assert lease is not None + + with session_factory() as session: + for _ in range(3): + assert plane.cancel_requested(session, job_id) is True + assert plane.renew(session, lease).cancel_requested is True + + with session_factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.cancel_requested is True + assert job.status == "running" + + +# -- the in-process compatibility runner ------------------------------------- + + +class _RecordingWorker: + """A worker stand-in that records how the runner drives it.""" + + def __init__(self, outcomes: list[bool]) -> None: + self.worker_id = "worker-recording" + self._outcomes = list(outcomes) + self.run_once_calls = 0 + self.sweep_calls = 0 + self.shutdown_calls = 0 + self.drained = threading.Event() + + def run_once(self) -> bool: + self.run_once_calls += 1 + if self._outcomes: + return self._outcomes.pop(0) + self.drained.set() + return False + + def sweep_stale(self) -> int: + self.sweep_calls += 1 + return 0 + + def shutdown(self) -> None: + self.shutdown_calls += 1 + + +def _runner(worker, **kwargs) -> AnalysisWorkerRunner: + kwargs.setdefault("poll_interval_seconds", 0.01) + return AnalysisWorkerRunner(worker, **kwargs) + + +def test_the_runner_sweeps_once_before_it_starts_polling(session_factory): + worker = _RecordingWorker([]) + runner = _runner(worker) + + runner.sweep_on_start() + + assert worker.sweep_calls == 1 + assert worker.run_once_calls == 0 + + +def test_the_runner_drains_a_backlog_without_sleeping_between_jobs(): + worker = _RecordingWorker([True, True, True]) + runner = _runner(worker, poll_interval_seconds=30) + + runner.start() + try: + assert worker.drained.wait(timeout=5), "the runner did not drain the backlog" + finally: + runner.stop(timeout=5) + + # Three claimed jobs, then the empty poll that set ``drained``. A runner + # that slept between jobs could not have reached the fourth call with a + # 30-second poll interval. + assert worker.run_once_calls >= 4 + + +def test_the_runner_sweeps_stale_jobs_on_its_configured_cadence(): + worker = _RecordingWorker([]) + runner = _runner(worker, stale_sweep_interval_polls=2) + + runner.start() + try: + assert worker.drained.wait(timeout=5) + _wait_until(lambda: worker.sweep_calls >= 2, timeout=5) + finally: + runner.stop(timeout=5) + + # One startup sweep plus at least one periodic sweep from the loop. + assert worker.sweep_calls >= 2 + + +def test_a_failing_iteration_does_not_kill_the_runner_loop(): + class _ExplodingWorker(_RecordingWorker): + def run_once(self) -> bool: + self.run_once_calls += 1 + if self.run_once_calls == 1: + raise RuntimeError("boom") + self.drained.set() + return False + + worker = _ExplodingWorker([]) + runner = _runner(worker) + + runner.start() + try: + assert worker.drained.wait(timeout=5), "the loop died on the first failure" + finally: + runner.stop(timeout=5) + + +def test_stopping_the_runner_signals_the_worker_and_joins_the_thread(): + worker = _RecordingWorker([]) + runner = _runner(worker) + + runner.start() + assert worker.drained.wait(timeout=5) + runner.stop(timeout=5) + + assert worker.shutdown_calls == 1 + assert runner._thread is None + assert not any(thread.name == "analysis-worker" for thread in threading.enumerate()) + + +def test_stopping_a_runner_that_never_started_is_safe(): + worker = _RecordingWorker([]) + runner = _runner(worker) + + runner.stop(timeout=1) + + assert worker.shutdown_calls == 1 + assert worker.run_once_calls == 0 + + +def test_starting_a_running_runner_twice_is_refused(): + worker = _RecordingWorker([]) + runner = _runner(worker) + + runner.start() + try: + with pytest.raises(RuntimeError): + runner.start() + finally: + runner.stop(timeout=5) + + +def test_worker_ids_are_unique_within_one_process(): + """Every ownership guard is ``worker_id`` equality, so collisions are fatal.""" + + ids = {new_worker_id(pid=7) for _ in range(50)} + + assert len(ids) == 50 + assert all(worker_id.startswith("analysis-worker-7-") for worker_id in ids) + assert all(len(worker_id) <= 64 for worker_id in ids) + + +# -- the worker still reaches the queue only through the boundary ------------ + + +def test_the_worker_claims_through_its_injected_control_plane(session_factory): + """The seam is real: swapping the control plane changes what the worker gets.""" + + from app.workers.analysis_worker import AnalysisWorker + + _bootstrap(session_factory) + + class _EmptyQueue(DatabaseAnalysisControlPlane): + def claim(self, session, *, worker_id): + return None + + worker = AnalysisWorker( + session_factory, + worker_id="worker-a", + lease_seconds=60, + control_plane=_EmptyQueue(lease_seconds=60), + ) + + assert worker.run_once() is False + + with session_factory() as reader: + assert reader.scalar(select(func.count()).select_from(AnalysisJob).where(AnalysisJob.status == "queued")) == 1 + + +def test_the_worker_defaults_to_the_database_control_plane(session_factory): + from app.workers.analysis_worker import AnalysisWorker + + worker = AnalysisWorker(session_factory, worker_id="worker-a", lease_seconds=60) + + assert isinstance(worker.control_plane, DatabaseAnalysisControlPlane) + assert worker.control_plane.lease_seconds == 60 + + +# -- PostgreSQL: the deployment authority ------------------------------------ + + +def _assert_threaded_claim_race_has_one_winner(factory) -> None: + """Two real concurrent claims, released together by a barrier.""" + + _, job_id = _bootstrap(factory) + barrier = threading.Barrier(2) + results: dict[str, object] = {} + errors: list[BaseException] = [] + + def claim(worker_id: str) -> None: + session = factory() + try: + # Both workers resolve their candidate before either writes, then + # the barrier releases them into the real race window still holding + # it -- so both genuinely reach the compare-and-swap. + candidate = _plane().next_eligible_job_id(session) + assert candidate == job_id + plane = _PinnedCandidatePlane(candidate, lease_seconds=60) + barrier.wait(timeout=10) + results[worker_id] = plane.claim(session, worker_id=worker_id) + except BaseException as exc: # noqa: BLE001 - surfaced to the assertion + errors.append(exc) + finally: + session.close() + + threads = [threading.Thread(target=claim, args=(f"worker-{name}",)) for name in ("a", "b")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + + assert not errors, errors + assert all(not thread.is_alive() for thread in threads) + winners = [worker_id for worker_id, lease in results.items() if lease is not None] + assert len(winners) == 1, f"expected exactly one winner, got {winners}" + + with factory() as reader: + job = reader.get(AnalysisJob, job_id) + assert job.status == "running" + assert job.worker_id == winners[0] + assert job.attempt == 1 + + +@pytest.mark.skipif(not PG_URL, reason="set PARTHA_TEST_PG_URL to run the Postgres control-plane concurrency test") +def test_concurrent_claims_have_exactly_one_winner_on_postgres(): + """PostgreSQL is the deployment authority for this race. + + SQLite serialises writers, so its version of this race is expressed as an + explicit interleaving above. Only a real MVCC server exercises two claims + genuinely in flight at once. + """ + + engine = create_engine(PG_URL) + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine, expire_on_commit=False) + try: + _assert_threaded_claim_race_has_one_winner(factory) + finally: + engine.dispose() + + +def _wait_until(predicate, *, timeout: float) -> None: + """Poll ``predicate`` until true, failing the test if it never becomes true.""" + + deadline = datetime.now(UTC) + timedelta(seconds=timeout) + while datetime.now(UTC) < deadline: + if predicate(): + return + threading.Event().wait(0.01) + raise AssertionError("condition was never met") From f2a98ba43ef649eef999f2d58c647b94139bcedc Mon Sep 17 00:00:00 2001 From: Shaurya K Sharma Date: Sat, 29 Aug 2026 22:28:58 +0100 Subject: [PATCH 4/4] docs(scale): document the analysis control-plane boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Record what now runs: the control plane and runner as separate component rows, the claim/lease exchange in the ingestion sequence, and the fact that the API process hosts a worker without owning the queue. The known-limits entry is widened rather than softened — analysis is still whole-repository and still runs one in-process worker per API process. No standalone worker deployment exists, and this documents current behaviour only. Refs #324, #210 --- docs/architecture/SYSTEM_OVERVIEW.md | 35 +++++++++++++++++++++------- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/docs/architecture/SYSTEM_OVERVIEW.md b/docs/architecture/SYSTEM_OVERVIEW.md index 538c04f9..8e9958f2 100644 --- a/docs/architecture/SYSTEM_OVERVIEW.md +++ b/docs/architecture/SYSTEM_OVERVIEW.md @@ -10,8 +10,9 @@ Audience: contributors and maintainers who need to know what runs, where the bou PARTHA is a monorepo: a React frontend, a FastAPI backend, and a local filesystem plus relational database for persistence. Durable analysis uses the -database as its queue and a daemon worker thread inside the API process; there -is no external message queue or separate analysis service. +database as its queue behind an explicit control-plane boundary, driven by a +daemon worker thread inside the API process; there is no external message queue +or separate analysis service. ```mermaid flowchart LR @@ -22,13 +23,15 @@ flowchart LR MW["Middleware
rate limit · security headers
CORS · request ID"] Routes["Routes
app/api/routes/"] Services["Services
app/services/"] + Queue["Control plane
app/workers/control_plane.py
claim · lease · reclaim"] Worker["Analysis worker
app/workers/ · daemon thread"] Extract["Extractors
app/extraction/"] RI["Repository Intelligence
app/intelligence/
sealed read model"] Consumers["Consumers
analysis · graph · review · insights
documentation · ai · reports"] MW --> Routes --> Services - Services -->|"enqueue"| Worker + Services -->|"enqueue"| Queue + Queue -->|"lease"| Worker Worker --> Extract --> RI Services --> Consumers Consumers -->|"read only"| RI @@ -48,6 +51,7 @@ flowchart LR UI --> MW RI --> DB + Queue --> DB Worker --> DB Services --> Disk Services --> GH @@ -84,7 +88,9 @@ flowchart LR | `auth/` | Argon2 password hashing, HS256 access tokens, rotating refresh tokens with reuse detection. | — | | `core/` | Settings and validation, database engine, structured logging with redaction, request IDs, metrics, rate limiting, security headers. | — | | `storage/` | Local filesystem storage for uploads and extracted/cloned repositories. Enforces path safety on extraction. | — | -| `workers/` | Database-backed durable analysis execution, lease renewal, bounded retry, cancellation, and stale-job reconciliation. | Serve request-specific data or bypass owner-scoped API services. | +| `workers/control_plane.py` | The queue boundary: which jobs are eligible, and who owns them — claiming, lease renewal, expiry, reclaim, and every ownership guard. | Analyse a repository, or read anything but `analysis_jobs`. | +| `workers/runner.py` | The in-process runner: worker identity, the poll/sweep loop, and shutdown. The same loop a standalone worker process would run. | Contain queue policy of its own, or be imported by the API for anything but start/stop. | +| `workers/analysis_worker.py` | Execution of an *already-claimed* job: the extraction pipeline, snapshot sealing, bounded retry, cancellation and stale-job reconciliation. | Decide who owns a job, or serve request-specific data. | --- @@ -102,6 +108,7 @@ sequenceDiagram participant Repo as RepositoryService participant Store as LocalStorage participant Parser as RepositoryParser + participant Queue as Control plane participant Worker as AnalysisWorker participant RI as Extraction + Intelligence participant DB as Database @@ -118,16 +125,26 @@ sequenceDiagram UI->>API: POST /analysis/{id}/start API->>DB: insert queued analysis_jobs row API-->>UI: queued + job id - Worker->>DB: claim job + renew lease by stage + Worker->>Queue: claim + Queue->>DB: compare-and-swap queued -> running + lease + Queue-->>Worker: JobLease + Worker->>Queue: renew lease by stage (reports cancellation) Worker->>RI: extract and resolve repository facts Worker->>DB: persist and seal normalized ri.v1 snapshot ``` Clone/archive extraction and initial file-tree parsing run synchronously during import. `POST /analysis/{id}/start` durably enqueues the analysis and returns -immediately. A daemon worker thread in the API process claims jobs from the -database, reports progress at completed stage boundaries, seals the normalized -snapshot, and serves every product consumer from that immutable read model. +immediately. A daemon worker thread in the API process claims jobs *through the +control plane*, reports progress at completed stage boundaries, seals the +normalized snapshot, and serves every product consumer from that immutable read +model. + +The API process hosts that worker but does not own the queue. Claiming, leases, +expiry, reclaim and ownership guards live in `app/workers/control_plane.py`, and +the poll/sweep loop lives in `app/workers/runner.py`, so the same components +would drive a standalone worker process without changing the claim/lease +contract. Running analysis in a separate process is not implemented today. --- @@ -268,7 +285,7 @@ These are properties of the system as built, not a wish list. populates immutable normalized snapshots, and every product consumer requires the latest owner-scoped snapshot matching the current revision. Missing or stale snapshots return 404 without fallback. -4. **Analysis is whole-repository.** It runs in a durable, cancellable background job with bounded retry and stale-worker recovery, but incremental re-analysis is not implemented. Import extraction and file-tree parsing remain synchronous. +4. **Analysis is whole-repository, and its worker is still in-process.** It runs in a durable, cancellable background job with bounded retry and stale-worker recovery, claimed through an explicit control plane, but incremental re-analysis is not implemented and no standalone worker process is deployed — one API process runs one worker. Import extraction and file-tree parsing remain synchronous. 5. **The rate limiter trusts only the direct socket peer for unauthenticated requests.** `X-Forwarded-For` is deliberately ignored, so behind a reverse proxy every unauthenticated client shares one IP budget until a trusted-proxy allowlist is designed. Authenticated requests are keyed per user and unaffected. 6. **Dependency coverage is narrow.** Three manifest formats plus two lockfile formats (`package-lock.json`, `poetry.lock`), whose exact pins are recorded as resolutions on the same dependency identity rather than as direct edges. There is no transitive resolution and no vulnerability or outdated-version scanning. The API exposes explicit `not_computed` assessment statuses and does not emit a clean result or count without a scanner. 7. **Frontend assurance remains focused.** Vitest covers shared and feature