chore(scale): introduce the W2 queue and control-plane boundary by SHAURYAKSHARMA24 · Pull Request #376 · Second-Origin/PARTHA

SHAURYAKSHARMA24 · 2026-08-29T21:42:47Z

Summary

Establishes the queue/control-plane boundary for durable analysis jobs. Analysis jobs were already durable, leased, cancellable and crash-recoverable on dev — what was missing was a boundary: "who owns a job" existed only as private methods of AnalysisWorker, and the control loop that drove them lived inside the FastAPI application module. This PR extracts both, without changing the claim/lease/cancellation contract or any API surface.

Baseline


Starting `origin/dev`	`37dd2b5467ea9b0587e1944036f5a414ca586d60`
Branch	`chore/324-queue-control-plane-boundary`
Final head	`f2a98ba43ef649eef999f2d58c647b94139bcedc`

Linked issue

Closes #324
Refs #210

#210 stays open: it is the GATE B parent and remains so until #324–#327 are all complete.

Scope

Issue or RFC this advances: chore(scale): introduce the W2 queue and control-plane boundary #324 (child of chore(scale): W2 worker/queue/incremental analysis and recovery (GATE B) #210, roadmap §23 W2 / §24 GATE B).
Why this is in scope: it is operability-only infrastructure on an existing tracked issue. No new product surface, no new endpoint, no new language, no new parser. It deepens the existing durable-analysis moat by making worker ownership an explicit, reusable contract instead of an implementation detail of one class.
Accepted evidence it is real: 31 new tests over the boundary (all passing, including a threaded race against real PostgreSQL); the full backend suite unchanged apart from those additions; the generated API contract verified unchanged.

Dependency gate verified before starting

#210 says its children must not begin until GATE A / foundation prerequisites pass. Verified against code on dev, not issue state:

Prerequisite	Evidence on `dev`
Golden/benchmark corpus + drift checking	`apps/backend/tests/benchmark/` (`determinism.py`, `provenance.py`, `config/`, `fixtures/`)
Capability/support registry	`app/extraction/support_matrix.py` + `scripts/check-capabilities.py` (its own CI job)
Deterministic `ri.v1` facts/evidence	`app/intelligence/canonical.py`, `app/intelligence/snapshot_store.py`
Python + TypeScript/JS extraction	`app/extraction/python.py`, `app/extraction/typescript.py`
Lockfile facts	`app/extraction/lockfiles.py`
Service-interaction facts	`app/extraction/http.py`, consumed by both extractors as `http_call` observations
IaC facts	`app/extraction/iac.py`

All seven are registered in production_extractors() / wired into the sealed pipeline. #208 and #209 are closed. The gate is genuinely satisfied; nothing from #208/#209 is implemented here.

Existing functionality discovered (already present before this PR)

This matters so that pre-existing work is not misattributed to this PR. All of the following already worked on dev and were moved, not written:

durable DB-backed jobs and the queued/running/completed/failed/cancelled lifecycle;
compare-and-swap claiming with a status='queued' guard;
worker ids, leases, and stage-boundary lease renewal;
a heartbeat thread renewing the lease during a long stage, including the SQLite busy-timeout handling;
cooperative cancellation with RI-JOB-CANCELLED;
bounded retry with exponential backoff;
stale-worker reconciliation (sweep_stale) and sealed-snapshot self-healing;
an in-process daemon worker loop started from app.main.

Roughly two-thirds of #324's acceptance criteria were therefore already behaviourally satisfied. What was missing was structural, and that is what this PR delivers.

Problem actually solved

Two couplings, both structural:

The executor owned the queue. AnalysisWorker._claim, _heartbeat_once's renewal CAS, sweep_stale's expiry scan, _reconcile_stale's reclaim CAS and _update_owned_job's ownership predicate were private methods of the class that also runs the extraction pipeline. There was no lease type, no interface, and no way for a separate worker process to participate in the queue without importing the whole executor.
The API process owned the control loop. app.main minted the ownership token, constructed the worker, ran the poll loop, decided the stale-sweep cadence (_ANALYSIS_STALE_SWEEP_INTERVAL = 10) and joined the thread. Nothing outside a FastAPI lifespan could run a worker without copying that policy.

Implementation

Two new modules; no new runtime dependency.

app/workers/control_plane.py — the boundary.

AnalysisControlPlane (Protocol): next_eligible_job_id, claim, renew, expired_job_ids, reclaim, update_owned, cancel_requested.
DatabaseAnalysisControlPlane: the v1 implementation, backed by the existing analysis_jobs table.
JobLease is a frozen 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. deferred names the previously anonymous SQLite single-writer case: ownership is unchanged, so the pulse is skipped rather than misread as a lost lease.

app/workers/runner.py — the in-process compatibility path.

AnalysisWorkerRunner owns worker identity, the poll loop, the sweep cadence, and shutdown. run_forever() is a plain blocking call, so the loop a standalone worker process needs already exists rather than being trapped inside an async context manager.
new_worker_id() moved here from app.main, since worker identity is what every ownership guard compares.

AnalysisWorker keeps execution only: running a job it already owns through the ri.v1 pipeline and deciding its terminal transition. It takes an optional control_plane argument (defaulting to the database one), which is the seam a future backing store or worker process would use.

Why not Redis/Celery/RQ

Deliberately not added. The durable analysis_jobs row is already the authority for job identity, attempt budget, cancellation and lease expiry, and crash safety already comes from idempotent reconciliation rather than broker delivery semantics. A broker would introduce 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 is the abstraction that makes a different backing store possible later without redesigning the claim/lease contract.

Why `next_eligible_job_id` is public

A claim is a candidate read followed by a compare-and-swap. Splitting the read out names the half that carries no ownership, makes the race window visible to a reader, and — critically — makes it reachable by a test that must hold a stale candidate across another worker's claim. See "Concurrency evidence" below for why that mattered.

Concurrency model

Concern	Mechanism
Claim ownership	`UPDATE ... WHERE id = :id AND status = 'queued'`. Portable CAS, not `FOR UPDATE SKIP LOCKED`, so SQLite and PostgreSQL share one code path and one contract.
Duplicate-claim protection	The `status='queued'` predicate. Two workers racing one row see exactly one non-zero `rowcount`; the loser gets `None` and polls again, and does not spend an attempt.
Leases	`lease_expires_at`, set on claim and extended on renewal and at every stage checkpoint.
Renewals	`UPDATE ... WHERE id AND status='running' AND worker_id = <owner>` with `RETURNING cancel_requested`. Renewal and cancellation observation are one statement, so a request accepted between them cannot be missed.
Expiry	`expired_job_ids` is read-only and is not a claim; a caller must still win `reclaim` for each id.
Reclaim	The guard pins the exact prior owner and the exact lease instant and requires the lease to still be lapsed. A renewal landing between scan and reclaim moves `lease_expires_at`, so the CAS matches nothing and an active lease can never be stolen.
Ownership	`update_owned` refuses any mutation by a non-owner or on a non-running job. It neither commits nor rolls back — recovery is the caller's decision (abandon / cancel / reconcile).
Cancellation	Read from the row, never carried on the lease, so a request accepted after the claim is still seen. `require_cancel_not_requested` makes a completion that raced an accepted cancellation lose, which is what keeps cancellation idempotent and non-discardable.
Stale handling	Unchanged: reclaim, then heal-to-completed if a snapshot was sealed, else fail the orphaned building snapshot and requeue with backoff or fail terminally at the attempt limit.

Reclaim does not spend an attempt. Attempt budget is retry policy (#325); a handoff is not a retry. Asserted explicitly in test_an_expired_lease_is_reclaimable_by_another_worker.

Two-commit boundary preserved

The intentional snapshot-seal / job-completion two-commit boundary is untouched, and the module note forbidding its collapse is retained and strengthened. Crash safety still comes from idempotent stale reconciliation, not from a transaction rewrite.

Concurrency evidence: a mutation check that found a bad test

The first version of test_two_workers_racing_one_job_produce_exactly_one_owner passed even with the status='queued' guard deleted — because on SQLite the loser's internal candidate read ran after the winner committed, so it returned None before ever reaching the CAS. The test was asserting the right invariant for the wrong reason.

Fixed by pinning the candidate each worker read (_PinnedCandidatePlane) so both genuinely reach the swap. Re-verified by mutation: deleting the guard now fails two tests. The threaded PostgreSQL test uses the same pinning behind a threading.Barrier.

Compatibility

The in-process single-worker path is preserved exactly, as chore(scale): introduce the W2 queue and control-plane boundary #324 requires during migration. ANALYSIS_WORKER_AUTOSTART behaves identically.
app.main no longer contains queue policy; it starts and stops a runner.
AnalysisWorker._claim(session) is retained as a row-shaped delegation so existing callers and tests are unchanged.
A standalone worker process would build the same worker and call runner.run_forever(), adding no new queue policy. Building that deployment is not in this PR — it belongs to chore(scale): W2 worker/queue/incremental analysis and recovery (GATE B) #210.

Acceptance criteria completed

AC	Status	Evidence
AC1 Jobs claimed through an explicit lease/control-plane path	Complete	`DatabaseAnalysisControlPlane.claim` returns a `JobLease`; `AnalysisWorker.run_once` obtains it via `self.control_plane`. Tests: `test_claim_returns_a_lease_and_marks_the_job_running`, `test_claim_skips_a_job_still_serving_retry_backoff`, `test_claim_takes_the_oldest_eligible_job_first`, `test_the_worker_claims_through_its_injected_control_plane`
AC2 Duplicate claims and expired leases handled safely	Complete	`test_two_workers_racing_one_job_produce_exactly_one_owner`, `test_a_second_claim_of_a_running_job_takes_nothing`, `test_concurrent_claims_have_exactly_one_winner_on_postgres`, `test_an_expired_lease_is_reclaimable_by_another_worker`, `test_an_active_lease_is_never_stolen`, `test_a_renewal_between_scan_and_reclaim_defeats_the_reclaim`
AC3 Cancellation observable and idempotent	Complete	`test_a_renewal_reports_a_cancellation_request`, `test_a_cancellation_request_survives_a_reclaim_handoff`, `test_cancellation_is_not_resurrected_after_a_cancelled_job_is_reclaimed`, `test_the_cancel_not_requested_guard_refuses_to_complete_a_cancelling_job`, `test_repeated_cancellation_reads_are_idempotent`, plus the unchanged `test_analysis_worker.py` cancellation suite
AC4 Existing analysis behaviour and API contracts intact	Complete	Full backend suite: same 9 pre-existing failures as baseline, no new ones. API contract check reports no drift. Frontend lint/test/build green. No model, migration, route or schema change.
AC5 Tests cover ownership, lease expiry and retry boundaries	Complete	Ownership: `test_a_non_owner_cannot_mutate_another_workers_job`, `test_a_non_owner_cannot_renew_another_workers_lease`, `test_a_worker_that_lost_its_lease_is_told_so_on_the_next_renewal`, `test_a_terminal_job_cannot_be_renewed_or_mutated`. Expiry: the AC2 reclaim tests + `test_lease_expired_compares_naive_and_aware_timestamps`. Retry boundary: `test_claim_skips_a_job_still_serving_retry_backoff` and the retained `test_analysis_worker.py` bounded-retry/sweep tests.

Testing performed

# Backend, SQLite (default local configuration)
$ python -m pytest
1113 tests: 1088 passed, 9 failed, 16 skipped

# Backend, with a real PostgreSQL 16 server
#   PARTHA_TEST_PG_URL=postgresql+psycopg://...@127.0.0.1:55432/partha_test
$ python -m pytest
1113 tests: 1100 passed, 9 failed, 4 skipped

# The new boundary suite, against real PostgreSQL (no skips)
$ python -m pytest tests/test_analysis_control_plane.py
31 passed

# Static analysis
$ ruff check app                 -> All checks passed!
$ ruff format --check app        -> 143 files already formatted
$ mypy app                       -> Success: no issues found in 143 source files
$ python scripts/check-capabilities.py
  -> Capability registry valid: 62 benchmark mappings
  -> README capability registry is current and deterministic

# Frontend / contract
$ npm --prefix apps/frontend run generate:api-contract -- --check
  -> API contract is up to date
$ npm run lint:frontend          -> clean
$ npm --prefix apps/frontend run test  -> 74 files, 439 tests passed
$ npm run build:frontend         -> built in 31.97s

# Repository hygiene
$ git diff --check               -> no whitespace errors

Baseline comparison (this is why the 9 failures are not regressions). The identical 9 tests fail on unmodified origin/dev 37dd2b5 in this environment:

origin/dev 37dd2b5 : 1082 tests, 1058 passed, 9 failed, 15 skipped
this branch        : 1113 tests, 1088 passed, 9 failed, 16 skipped
delta              : +31 tests, +30 passed, +1 skipped, same 9 failures

All 9 are Windows symlink-privilege failures (os.symlink needs elevation), e.g. test_repository_parser_rejects_a_symlink_that_escapes_the_checkout. They are environmental and unrelated to this change; they pass on CI's Linux runners.

Not run locally: npm run test:e2e and npm run test:accessibility (Playwright browser journeys). No frontend production file changed and no route, schema or page structure was touched, so no journey is affected — but I did not run them, and I am relying on CI for those two jobs rather than claiming them.

Migrations

No migration required. The analysis_jobs schema already carries every column the control plane needs — worker_id, lease_expires_at, next_attempt_at, attempt, max_attempts, cancel_requested — plus ix_analysis_jobs_status_lease, which already indexes exactly the expiry scan. git diff over app/models/ and alembic/ is empty.

Dependencies and blocked work

No new runtime or dev dependency (pyproject.toml, requirements-dev.txt and both package.json files are unchanged). Depends on nothing; blocks nothing.

Screenshots

Not applicable — no user-visible change.

Security and data considerations

Owner scoping is unchanged. The control plane operates on the queue, which is worker-scoped by design; every owner-scoped read/write path (AnalysisJobService) is untouched.
No new logging of repository contents, credentials or secrets. The new log lines carry only job_id, worker_id and a timeout value.
Ownership guards are the security-relevant surface here: a non-owner must not mutate, renew, complete or cancel another worker's job. That is covered by four explicit tests and verified by mutation testing on the claim guard.
No new endpoint, no new input from an untrusted source, no migration, no data loss path.

Scope exclusions

Explicitly not implemented by this PR:

chore(scale): add worker failure recovery and bounded retry semantics #325 — worker failure-recovery redesign and bounded retries beyond the existing behaviour this PR preserves.
chore(scale): add content-cache invalidation keyed by canonical graph hash #326 — content cache and invalidation.
chore(scale): prove full and incremental analysis convergence #327 — full/incremental convergence and budget evidence.
feat: two-revision graph-diff spike for evidence-backed change-impact wedge #219 — graph diff.
Incremental analysis itself, change-impact surfaces, new languages, vulnerability scanning, auth work, UI changes, deployment-platform changes, telemetry, and unrelated cleanup.

Scope changes or remaining work

No scope change. Genuine remaining limitations, recorded honestly:

Still one in-process worker per API process. This PR makes a standalone worker possible without another redesign; it does not deploy one. That remains chore(scale): W2 worker/queue/incremental analysis and recovery (GATE B) #210's work, and SYSTEM_OVERVIEW.md's known-limits entry says so plainly rather than implying distribution now exists.
reclaim takes a live ORM object. The production caller (sweep_stale) re-reads a clean row first, so the pinned guard values are correct. A caller passing a dirty job object could autoflush stale values into the row before the guard runs. Not a live defect; worth tightening to a plain value object if a second sweeper caller ever appears.
The stale-sweep cadence is still poll-count based (every 10 empty polls), inherited unchanged from app.main. It is now a runner constructor argument rather than a module constant, so chore(scale): add worker failure recovery and bounded retry semantics #325 can make it time-based without touching the boundary.

Unrelated observation, not fixed here (out of scope, no issue filed): apps/backend/nul is a stray untracked 39-byte file dating from 30 July, an artifact of a Windows shell redirect to nul. It is not included in this PR and is not mine, but it will keep showing up in git status for anyone on Windows until someone deletes it or .gitignores it.

Contributor checklist

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 Second-Origin#324, Second-Origin#210

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 Second-Origin#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 Second-Origin#324 requires during migration. Building the standalone deployment stays with Second-Origin#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 Second-Origin#324, Second-Origin#210

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 Second-Origin#324, Second-Origin#210

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 Second-Origin#324, Second-Origin#210

SHAURYAKSHARMA24 added 4 commits August 29, 2026 22:28

SHAURYAKSHARMA24 requested a review from parthrohit22 as a code owner August 29, 2026 21:42

SHAURYAKSHARMA24 self-assigned this Aug 29, 2026

Navigation Menu

Uh oh!

chore(scale): introduce the W2 queue and control-plane boundary - #376

chore(scale): introduce the W2 queue and control-plane boundary#376
SHAURYAKSHARMA24 wants to merge 4 commits into
Second-Origin:devfrom
SHAURYAKSHARMA24:chore/324-queue-control-plane-boundary

SHAURYAKSHARMA24 commented Aug 29, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

SHAURYAKSHARMA24 commented Aug 29, 2026

Summary

Baseline

Linked issue

Scope

Dependency gate verified before starting

Existing functionality discovered (already present before this PR)

Problem actually solved

Implementation

Why not Redis/Celery/RQ

Why next_eligible_job_id is public

Concurrency model

Two-commit boundary preserved

Concurrency evidence: a mutation check that found a bad test

Compatibility

Acceptance criteria completed

Testing performed

Migrations

Dependencies and blocked work

Screenshots

Security and data considerations

Scope exclusions

Scope changes or remaining work

Contributor checklist

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Why `next_eligible_job_id` is public