UN-3546 [FEAT] Priority-ordered PG-queue dequeue + concurrency-safe claim by muhammad-ali-e · Pull Request #2052 · Zipstack/unstract

muhammad-ali-e · 2026-06-12T13:52:20Z

Starts enforcing the load-independent part of fairness — pipeline_priority (L3) — directly in the single-table dequeue. The fairness key already flowed on every PG message but was pure pass-through; now a higher-priority task is claimed first, FIFO within a priority. Targets feat/UN-3445-pg-queue-integration.

⚠️ Also fixes a latent concurrency bug in the core dequeue (9a)

While testing the new ordering, the claim over-claimed under concurrency — read(qty=1) returned 2 rows when other transactions touched the table. Root cause is in the original dequeue, not the new ordering:

UPDATE … WHERE msg_id IN (SELECT … FOR UPDATE SKIP LOCKED LIMIT n) — EvalPlanQual re-evaluates the LIMIT subquery when a row it tried to lock was concurrently modified, so a single claim can return more than n rows. With concurrent consumers that means a read(qty=1) could pull (and bump the vt on) extra messages.

Fix — the canonical PGMQ-safe shape: lock candidates in a CTE, then UPDATE … FROM locked WHERE q.msg_id = locked.msg_id (locks exactly n once). The trailing SELECT re-orders RETURNING (otherwise unspecified) so batched claims also come back in priority order.

WITH locked AS (
    SELECT msg_id FROM pg_queue_message
     WHERE queue_name = %s AND vt <= now()
     ORDER BY priority DESC, msg_id
     FOR UPDATE SKIP LOCKED LIMIT %s
), claimed AS (
    UPDATE pg_queue_message q SET vt = now() + …, read_ct = read_ct + 1
      FROM locked WHERE q.msg_id = locked.msg_id
    RETURNING q.msg_id, q.message, q.read_ct, q.priority
)
SELECT msg_id, message, read_ct FROM claimed ORDER BY priority DESC, msg_id

This benefits the whole queue (every caller), not just priority ordering.

What

Schema (models.py + migration 0002): priority smallint default 5 (= FairnessKey.DEFAULT_PRIORITY); dequeue index swapped to (queue_name, priority DESC, msg_id) so the priority claim stays an indexed top-N (no full-backlog sort).
Enqueue: dispatch() writes priority from fairness.pipeline_priority; a bare dispatch (fairness=None) writes the neutral default.
Dequeue: ORDER BY priority DESC, msg_id (higher first, FIFO within a priority).

Deliberately deferred (need load / belong with the orchestrator)

L1 org-tier + L2 workload ordering, org_config/burst_max admission, staging_queue/task_queue split, orchestrator_lock leader election — the multi-org fair-admission orchestrator, justified once the pipeline routes cross-org load through PG.

Testing

Integration (real Postgres): priority selection one-at-a-time + batch ordering; 48 tests pass 5× consecutively across client+dispatch+routing (the combo that exposed the over-claim).
Unit: send() writes priority; dispatch wiring (fairness + neutral default); read param order.
Live end-to-end: dispatched 5 mixed-priority tasks via the real dispatch() path → drained in claim order 9 > 7 > 3a > 3b(FIFO) > 1.
Real flow: live API/ETL workflow notifications land in pg_queue_message with priority: 5 (neutral, fairness=null) — column populated end-to-end through the backend producer.
Migration applies clean; descending index confirmed in the DB.

🤖 Generated with Claude Code

…laim Start enforcing the load-independent part of fairness — pipeline_priority (L3) — directly in the single-table dequeue: higher priority is claimed first, FIFO (msg_id) within a priority. The org-tier (L1) / workload (L2) axes + burst_max admission stay deferred to the fair-admission orchestrator. - Schema: add `priority` (smallint, default 5 = FairnessKey.DEFAULT_PRIORITY) to pg_queue_message; swap the dequeue index to (queue_name, priority DESC, msg_id) so the priority-ordered claim stays an indexed top-N. - Enqueue: dispatch() writes priority from fairness.pipeline_priority; a bare dispatch (fairness=None) writes the neutral default. - Dequeue: ORDER BY priority DESC, msg_id. Also fixes a latent concurrency bug in the original dequeue (9a): `UPDATE ... WHERE msg_id IN (SELECT ... FOR UPDATE SKIP LOCKED LIMIT n)` can OVER-CLAIM under concurrent writers — EvalPlanQual re-evaluates the LIMIT subquery when a row it tried to lock was concurrently touched, so one claim can return more than n rows. Switched to the canonical PGMQ-safe shape: lock candidates in a CTE, then `UPDATE ... FROM locked WHERE q.msg_id = locked.msg_id`, which locks exactly n rows once. The trailing SELECT re-orders RETURNING (which is otherwise unspecified) so batched claims come back in priority order too. Tests: priority selection (one-at-a-time) + batch ordering against real Postgres; send writes priority; dispatch wiring (fairness + neutral default); read param order. Verified live end-to-end via dispatch()->claim-order (9>7>3a>3b>1). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

coderabbitai · 2026-06-12T13:52:28Z

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 18086c96-5357-48c1-86f7-5fe15523004f

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests
Commit unit tests in branch UN-3546-pg-queue-priority-dequeue

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

muhammad-ali-e

PR Review Toolkit — automated multi-agent review

Ran six specialized reviewers (Code Reviewer, Silent-Failure Hunter, Type-Design Analyzer, Test Analyzer, Comment Analyzer, Code Simplifier) over the diff vs feat/UN-3445-pg-queue-integration. Inline comments are attached below; this is the prioritized summary.

The PR is solid — the CTE FROM-join claim (vs the EvalPlanQual-unsafe IN (SELECT … LIMIT n) form) is the correct PGMQ-safe shape, param ordering is right, the migration sequence is valid, and the _enqueue_pg re-raise-with-breadcrumb (no silent Celery fallback) is exactly right. Findings are about hardening, not bugs in the happy path.

🔴 Critical / Important

priority has no range enforcement (flagged independently by 4 of 6 agents). client.send() writes any int straight to the DB, and the priority column has no CHECK — only FairnessKey.__post_init__ validates [1,10], and that sits on just one of the write paths. An out-of-range value is silently durable and corrupts dequeue ordering with no log/error. Today's only caller passes a validated value, but this PR is the plumbing that lets an arbitrary priority through. Fix at both layers: a ValueError guard in send() (mirroring its existing vt_seconds/qty guards) and a DB CheckConstraint (the only backstop for the raw-SQL/ORM writer). See inline on client.py send() and models.py.
Index dropped vt; the "indexed top-N" comment over-claims. The new (queue_name, priority DESC, msg_id) index removes vt, so visibility is a per-row filter. Under at-least-once delivery, in-flight (future-vt) high-priority rows pile up at the front of the index and get scanned-and-skipped on every claim — per-claim cost grows with the in-flight backlog, not LIMIT. Soften the comment and consider a partial index / benchmark with a realistic in-flight backlog. See inline on client.py:57 and models.py:43.

🟡 Test gaps (inline on `test_pg_queue_client.py:283`)

The rewrite's whole reason for existing — CTE prevents over-claim under concurrent writers — has zero coverage (test_no_double_delivery only proves disjointness, not the ≤qty bound).
vt × priority interaction is untested (a future-vt high-priority row should be skipped for a visible lower-priority one).
FIFO-within-band for batch claims only exercises single-member bands.

🟢 Comment / cleanup nits (inline)

Module docstring (client.py:8-10, outside the diff so not inline-commentable) still describes the old UPDATE … WHERE msg_id IN (SELECT … FOR UPDATE SKIP LOCKED …) shape this PR replaced and argues against — please update it to the CTE form.
models.py:34: FairnessKey.DEFAULT_PRIORITY is not a real symbol (it's module-level fairness.DEFAULT_PRIORITY).
client.py:76: param-order comment is duplicated at the call site (224-225) — drop one.

Generated with Claude Code via the PR Review Toolkit.

- Validate priority at the write boundary: client.send() raises ValueError on out-of-range (mirrors its vt_seconds/qty guards) — an out-of-range value would silently jump/sink the row in the priority DESC claim order. - Add a DB CheckConstraint (priority 1..10) as the backstop no ORM/raw writer can bypass (migration 0003). check= (not condition=) — repo is on Django 4.2. - Soften the "indexed top-N" comments (client.py + models.py): the dequeue is an index walk with vt<=now() as a per-row filter, NOT a guaranteed top-N — vt is not in the index, so in-flight (future-vt) high-priority rows are scanned past on each claim; the orchestrator's admission is the high-backlog answer. Update the module docstring to the CTE FROM-join shape; fix the fairness.DEFAULT_PRIORITY symbol reference; drop the duplicated param-order comment. - Tests: send() range-guard (parametrized) + DB CheckConstraint backstop; concurrent-writer over-claim guard (two readers, no batch exceeds qty — the regression test for the EvalPlanQual fix); vt × priority (visible low beats invisible high); FIFO-within-band for multi-member batch bands. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

muhammad-ali-e · 2026-06-12T14:09:01Z

Review feedback addressed — `15d1d45cf`

Thanks — all findings handled (threads replied inline).

🔴 Important

priority validation — send() now raises on out-of-range (mirrors its vt_seconds/qty guards), plus a DB CheckConstraint(1..10) (migration 0003) as the backstop no ORM/raw writer can bypass. (check=, since the repo is Django 4.2; condition= is 5.1+.)
index "top-N" overclaim — softened both comments: it's an index walk with vt <= now() as a per-row filter, not a guaranteed top-N; in-flight high-priority rows are scanned past (cheap at low depth; orchestrator admission is the high-backlog answer). A vt <= now() partial index isn't possible (now() is STABLE), so I documented the limit; benchmarking under a realistic backlog noted as follow-up.

🟡 Tests — added all three: concurrent-writer over-claim guard (the regression test for the EvalPlanQual fix), vt × priority (visible low beats invisible high), and FIFO-within-band for multi-member batch bands. Plus the send() range-guard + DB-constraint backstop tests.

🟢 Nits — module docstring updated to the CTE FROM-join shape; fairness.DEFAULT_PRIORITY symbol fixed; duplicate param-order comment dropped.

Verified: 56 tests pass 3× consecutively (client + dispatch + routing); migration 0003 applies clean.

greptile-apps · 2026-06-12T14:20:17Z

Greptile Summary

This PR enforces pipeline_priority (fairness L3) in the single-table PG-queue dequeue and simultaneously fixes a latent concurrency bug in the original claim logic. The dequeue index is replaced with (queue_name, priority DESC, msg_id) and _DEQUEUE_SQL is rewritten to the EvalPlanQual-safe CTE form that locks exactly qty rows once, preventing the over-claim that the old UPDATE … WHERE msg_id IN (SELECT … FOR UPDATE SKIP LOCKED LIMIT n) could produce under concurrent writers.

Schema + migrations: priority smallint default 5 added in 0002, CheckConstraint [1,10] in 0003, dequeue index updated. Both migrations replay cleanly from scratch.
Enqueue: dispatch() now passes fairness.pipeline_priority (or DEFAULT_PRIORITY) to PgQueueClient.send(), which validates the range before the DB insert; the CheckConstraint is the backstop for any raw writer.
Dequeue / concurrency fix: The CTE form (locked → claimed → outer ORDER BY) correctly selects the top-priority visible rows, updates exactly those rows, and returns results in a stable priority+FIFO order for both qty=1 and batch claims.

Confidence Score: 5/5

Safe to merge. The CTE rewrite directly addresses a real over-claim defect that was reproducible under concurrent load, and the priority ordering is a straightforward additive column with a DB-level check constraint as a backstop.

All changes are behaviorally correct and thoroughly tested — 48 integration tests including priority ordering, batch FIFO, vt×priority interaction, and a concurrent two-reader drain test that specifically exercises the over-claim fix. Schema, application-level guards, and DB constraints are mutually reinforcing. The SQL CTE shape matches the canonical PGMQ-safe dequeue pattern. No functional regressions were identified.

No files require special attention. workers/queue_backend/pg_queue/client.py carries the most weight (the dequeue SQL rewrite and parameter order change) but is validated by both unit assertions on the exact param tuple and live integration tests.

Important Files Changed

Filename	Overview
workers/queue_backend/pg_queue/client.py	Core change: rewrites _DEQUEUE_SQL to the EvalPlanQual-safe CTE form (lock in `locked`, UPDATE via FROM-join in `claimed`), adds ORDER BY priority DESC, msg_id, and updates send() to write and validate priority. Parameter order correctly updated to match new SQL %s positions.
backend/pg_queue/models.py	Adds priority SmallIntegerField(default=5), the CheckConstraint in Meta.constraints, and updates the dequeue index to (queue_name, priority DESC, msg_id). In sync with both migrations.
backend/pg_queue/migrations/0002_remove_pgqueuemessage_pg_queue_message_dequeue_idx_and_more.py	Removes the old (queue_name, vt, msg_id) index, adds priority smallint default 5, and recreates the dequeue index as (queue_name, priority DESC, msg_id). Correct Django 4.2 migration; existing rows receive the neutral default 5.
backend/pg_queue/migrations/0003_pgqueuemessage_pg_queue_message_priority_range.py	Adds CheckConstraint priority BETWEEN 1 AND 10 in a separate migration. The comment correctly flags the check= to condition= rename needed when Django is upgraded to >= 6.0.
workers/queue_backend/dispatch.py	Threads fairness.pipeline_priority (or DEFAULT_PRIORITY for bare dispatches) through to PgQueueClient.send(). Clean and minimal change.
workers/tests/test_pg_queue_client.py	Adds comprehensive unit (priority write/reject) and integration tests (priority ordering, batch FIFO, vt×priority interaction, concurrent claim safety, and constraint drift guard). The worker termination assertion and uuid-based queue name fix are both present.
workers/tests/test_dispatch_pg.py	Adds TestDispatchPriorityWiring with two mocked tests verifying priority flows from fairness key and uses DEFAULT_PRIORITY for bare dispatches.

Sequence Diagram

sequenceDiagram
    participant D as dispatch()
    participant C as PgQueueClient.send()
    participant DB as pg_queue_message (DB)
    participant R as PgQueueClient.read()
    participant W as Consumer Worker

    D->>C: send(queue, payload, org_id, priority)
    Note over C: validate priority in [MIN,MAX]
    C->>DB: "INSERT (queue_name, message, org_id, priority, vt=now())"
    DB-->>C: msg_id

    W->>R: read(queue, vt_seconds, qty)
    R->>DB: "WITH locked AS (SELECT msg_id WHERE queue_name=? AND vt<=now() ORDER BY priority DESC, msg_id FOR UPDATE SKIP LOCKED LIMIT ?)"
    Note over DB: Locks exactly qty rows, skips contended rows
    DB->>DB: "UPDATE FROM locked SET vt=now()+vt_seconds, read_ct+=1 RETURNING"
    DB-->>R: (msg_id, message, read_ct) ORDER BY priority DESC, msg_id
    R-->>W: [QueueMessage, ...]

    W->>DB: "DELETE WHERE msg_id=? (ack)"

_{Reviews (3): Last reviewed commit: "UN-3546 [DOCS] Note check=/condition= ha..." | Re-trigger Greptile}

- Concurrency test: assert the drain worker terminated after join (a hung worker now fails the test instead of passing silently while conn_b.close() races its in-flight queries). - Priority-bounds drift guard: backend models.py and workers fairness.py are separate codebases that can't import each other, so the DB constraint bounds (1/10) duplicate fairness.MIN/MAX_PRIORITY. Replaced the hardcoded "42" reject test with test_db_check_constraint_matches_fairness_bounds — raw-inserts at MIN/MAX (accepted) and MIN-1/MAX+1 (CheckViolation), pinning the DB constraint to the fairness range so a future widening that misses one side fails loudly. Documented the canonical source in the model comment. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

muhammad-ali-e · 2026-06-12T14:31:58Z

Greptile feedback addressed — `b68ec9fb4`

Concurrency test liveness — assert not worker.is_alive() after the join, so a hung drain fails loudly instead of passing silently (and racing conn_b.close()).
Priority-bounds drift — replaced the hardcoded 42 reject test with test_db_check_constraint_matches_fairness_bounds: raw-inserts at MIN_PRIORITY/MAX_PRIORITY (accepted) and just outside (CheckViolation), all from fairness.py — so the DB constraint and the app's fairness range can't silently diverge. Model comment documents fairness.py as canonical. (A shared @unstract/ constant is the proper long-term fix; offered as a follow-up since backend/workers can't import each other today.)

10 integration tests green; both findings were non-blocking (Greptile 4/5, safe-to-merge).

…ation Breadcrumb for a future Django upgrade: `check=` is correct on the pinned Django 4.2 (deprecated 5.1, removed 6.0); fresh installs always replay under the shipped Django, so leave it. When the pin reaches >= 6.0, squash (or do the behaviour-preserving check= -> condition= edit) so a from-scratch migrate runs. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

sonarqubecloud · 2026-06-12T14:39:14Z

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

* UN-3534 [FEAT] PG Queue Phase 8a — queue-transport routing gate + scaffold (#2033) * UN-3534 [FEAT] PG Queue Phase 8a — queue-transport routing gate + scaffold Add the Strangler-Fig routing seam that lets PG Queue (PGMQ) coexist with Celery so task types can be migrated one at a time. Scaffold only: the PG branch is a Celery-routing stub (no PG consumer exists yet), so this is zero-behaviour-change by construction. - queue_backend/routing.py: QueueBackend{CELERY,PG} + select_backend(task_name) reading the WORKER_PG_QUEUE_ENABLED_TASKS allow-list (default empty -> all Celery). Tolerant CSV parsing; never raises. - dispatch(): consults select_backend(); PG-selected tasks are logged but still dispatched via Celery. The send_task call sits outside the PG branch so the wire is byte-identical regardless of the routing decision. - queue_backend/pg_queue/: scaffold subpackage. PGMQ is a core transport substrate, so it lives in the seam beside dispatch/routing/barrier, not under the git-ignored plugins/ cloud overlay. - sample.env: documents the flag (default-safe, OSS-friendly, no Flipt server). - tests: 12 routing tests incl. the byte-identical-dispatch characterisation pinning the inert-scaffold invariant. Barrier axis untouched (WORKER_BARRIER_BACKEND stays chord). Per-org routing intentionally deferred to the rollout phase. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3534 [FEAT] Address Phase 8a review feedback - test seam: drop stale WORKER_PG_QUEUE_ENABLED_ORGS reference — the org axis was removed, that flag never existed [must-fix] - observability: routing log DEBUG -> INFO so a cutover survives a default log config; log-once per task name bounds volume. Log the configured allow-list once per process so a typo'd task name is eyeballable at boot even when it never matches a real dispatch [important] - tests: pin the routing branch with caplog assertions (PG -> log fires, Celery -> no log, bounded to once) so the inert gate can't be silently deleted; assert allow-list logging too [important — closes test gap] - QueueBackend: document the is-not-== discipline (StrEnum makes a typo'd "== cellery" a silent False) [suggestion] - routing: drop dead _parse_allow_list(env_var) param, read the constant directly; one-pass strip; test imports the constant (single source of truth) [nits] - pg_queue docstring: clarify plugins/ subdirs are git-ignored while the dir itself is tracked; soften volatile labs branch/section/filename references [nits] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3534 [DOCS] Note migration-coherence constraint in routing gate Document that the per-task allow-list may only split independent/leaf tasks across substrates. The coupled execution pipeline (async_execute_bin -> file processing -> callback, with the barrier fan-in) must run a single execution entirely on one transport — its migration unit is the execution, not the task. The next phase resolves transport once at kickoff and carries it in ExecutionContext; select_backend then honours that carried marker over the per-task env. Until then, only leaf tasks should be enabled here. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3534 [DOCS] Fix sample.env example to a leaf task (not pipeline) async_execute_bin is the pipeline kickoff — exactly the task the coherence note says must NOT be split per-task. Switch the example to a leaf task (send_webhook_notification) and warn against listing coupled pipeline tasks until ExecutionContext carries the transport choice. Addresses Greptile review feedback. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3534 [DOCS] Standardize on "PG Queue" naming; drop PGMQ branding We don't use the pgmq project (github.com/pgmq/pgmq) — no extension, no Python package, no copied SQL. The queue is a bespoke SKIP LOCKED schema (see the extension-free decision on UN-3533). Rename the 5 prose spots that called our substrate "PGMQ" to "PG Queue" so the code no longer reads as if it depends on that project. PGMQ stays only as prior-art reference in the decision record, not as the name of our substrate. Docs-only; no code or behaviour change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3537 [FEAT] PG Queue Phase 9a — extension-free queue schema + SKIP LOCKED dequeue (#2036) * UN-3537 [FEAT] PG Queue Phase 9a — extension-free queue schema + SKIP LOCKED dequeue The storage + dequeue primitive the routing seam will route to. Inert: nothing in dispatch() calls it yet (the PG branch still routes to Celery), so zero behaviour change even on the integration branch. Backend (schema only — SHARED_APPS, cross-org infra, shared schema): - new pg_queue app; 0001 is 100% makemigrations-generated (pg_queue_message table + dequeue index). No CREATE EXTENSION, no DB-side function — plain Django. managed=True model doubles as a typed read handle. Workers (the client; first direct-DB worker capability): - queue_backend/pg_queue: send / read / delete + QueueMessage. read() runs one atomic UPDATE ... FOR UPDATE SKIP LOCKED ... RETURNING (visibility- timeout pattern): claim+commit, process outside the txn, delete on success; a crash lets vt expire and the row redelivers (at-least-once, no double-delivery, VACUUM-safe, PgBouncer txn-pooling compatible). - connection.py reuses the backend DB_* env -> PgBouncer in cloud, direct in OSS (UN-3533 decision). - psycopg2-binary==2.9.9 (matches backend), promoted to a direct dep. Tests: 4 unit (mocked SQL shape) + 4 integration (real Postgres) proving roundtrip, vt-hiding, vt-expiry redelivery, and no-double-delivery. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3537 [FIX] org_id empty-string default, not null=True (Django S6553) A string-based field shouldn't have two "no data" values (NULL and ""). Use default="" for "no org" (leaf tasks) instead of null=True; regenerate the generated 0001 accordingly. The client coerces None -> "" since the column is now non-null. Addresses SonarCloud S6553. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3537 [FIX] Address Phase 9a review feedback Robustness + docstring-accuracy fixes on the PG-queue primitive: - client: roll back on error and recover a dead connection (drop the cached conn on OperationalError/InterfaceError -> next call reconnects); add close() + context manager. One connection blip no longer wedges the 9c consumer. - client.read(): raise on non-positive vt_seconds/qty (vt<=0 is a silent double-delivery window). - client.delete(): WARN when no row was removed (the at-least-once re-delivery signal was previously swallowed). - QueueMessage: slots=True + doc that the payload is decoded JSONB (the dict stays mutable; frozen freezes the binding only). - connection: parameterise create_pg_connection(env_prefix); wrap connect with a self-identifying error log (non-secret host/port/dbname/schema); drop the stale pg_queue_read reference; soften the DB_HOST default claim. - models: fix stale docstrings describing a removed 0002 / DB-level defaults / pg_queue_read function. - tests: narrow integration skip to OperationalError (don't mask ImportError/bugs/permission errors as a green skip); rollback before the teardown DELETE; integration conn delegates to create_pg_connection(env_prefix="TEST_DB_"); add unit coverage for create_pg_connection, read() validation, and error-rollback. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3537 [FIX] Address Phase 9a review round 2 + SonarCloud S2068 - perf: ORDER BY (vt, msg_id) so the (queue_name, vt, msg_id) index drives an indexed top-N instead of sorting the whole visible backlog on each read() — a real regression on a deep queue. - recovery: a failed rollback now proves the connection is dead, so a poisoned connection is recycled regardless of which psycopg2 error subclass was raised (not only Operational/Interface); also checks conn.closed. Closes the "one blip wedges the consumer" gap more fully. - docs: clarify the contract — at-least-once means a message CAN be processed more than once after vt-expiry; SKIP LOCKED only prevents concurrent double-claim. "(no double-delivery)" -> "(no concurrent double-claim)". - tests: cover the recovery/ownership branches (owned conn recycled on OperationalError and on failed rollback; injected conn never closed; close() owned-vs-injected). - security (SonarCloud S2068): the create_pg_connection test mapped a literal "PASSWORD": "p" -> use a runtime uuid token so it isn't flagged as a hard-coded credential. Resolves the Security Rating C gate failure. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3538 [FEAT] PG Queue Phase 9b — enqueue PG-routed tasks to Postgres (#2043) * UN-3538 [FEAT] PG Queue Phase 9b — enqueue PG-routed tasks to Postgres When a task is opted into WORKER_PG_QUEUE_ENABLED_TASKS, dispatch() now serialises it as a TaskPayload and enqueues to pg_queue_message instead of Celery, returning a PgDispatchHandle (.id = msg_id). A process-singleton PgQueueClient is reused across dispatches. The Celery path is unchanged and the default-empty flag routes everything to Celery — zero behaviour change. - queue_backend/pg_queue/task_payload.py: TaskPayload TypedDict + to_payload() — the producer<->consumer wire contract. This is the *contents* of the pg_queue_message.message JSONB column, distinct from the backend's PgQueueMessage *row* model (envelope vs payload — they nest, not duplicate). - dispatch(): PG branch enqueues + returns; cutover log (INFO, once per task); .warning docstring that an opt-in requires the 9c consumer running, else the task enqueues but never executes. - sample.env: same warning on the flag. - backend pg_queue model: note that `message` holds a TaskPayload. - tests: TestDispatchRouting (PG->enqueue, Celery->send_task), TestCutoverLog, to_payload shape + real-PG integration (dispatch lands a decodable row). Leaf-only (migration-coherence) — pipeline tasks stay on Celery until execution-level routing (9e). INERT by default. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3538 [FIX] Address Phase 9b review feedback - dispatch: per-thread PG client via threading.local — a libpq connection isn't safe for concurrent use across threads; correct under prefork AND a -P threads pool (gevent would need a pool — noted, out of scope). - dispatch: log the routing *decision* BEFORE send() so a first-dispatch failure (DB down / unmigrated) doesn't suppress the one announcement; wrap the enqueue with a logger.exception breadcrumb (a raw psycopg2.Error or a json.dumps TypeError on a non-serialisable arg otherwise propagate with no "PG-routed dispatch" context). No Celery fallback retained. - dispatch: hoist `pg_queue = queue or _DEFAULT_PG_QUEUE` (one source). - fairness: share a FairnessPayload TypedDict; to_dict() -> FairnessPayload; TaskPayload.fairness uses it instead of a loose dict[str, Any]. - pg_queue/__init__ docstring: describe 9b state (no longer "inert, rides Celery"); sample.env: drop a duplicated routes-to-Celery sentence. - tests: no-fallback-on-enqueue-failure (Critical gap) + log-ordering pin; lazy-init + reuse of the per-thread client; cutover-log assertions updated for the new message. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3539 [FEAT] PG Queue Phase 9c — consumer poll loop (claim → run → ack) (#2045) * UN-3539 [FEAT] PG Queue Phase 9c — consumer poll loop (claim → run → ack) PgQueueConsumer drains pg_queue_message and runs each claimed task in-process: poll_once() claims a batch (SKIP LOCKED + vt via PgQueueClient.read), runs it via current_app.tasks[name].apply(throw=True), and acks by deleting on success. Task failure -> leave the row (vt expiry redelivers, at-least-once); unknown task -> drop + error (no poison loop). run() adds an empty-queue backoff loop + SIGTERM/SIGINT graceful stop; main() is a `python -m` entrypoint (env-configured). Completes the leaf-first end-to-end path: 8a route -> 9b enqueue -> 9a store -> 9c consume+run. Validated live on the dev stack: a real send_webhook_notification routed to PG was claimed by the consumer, POSTed to Slack (HTTP 200), and acked (row removed). Deployment note: the consumer PROCESS must bootstrap the worker app (import the task modules) so current_app.tasks resolves the task — an entrypoint/rollout concern, not consumer logic. Documented in the module docstring; the rollout phase wires a consumer container that boots the app. Tests: 5 unit (run+ack, fail->no-ack, unknown->drop, empty, graceful stop) + 1 real-PG integration (enqueue -> poll -> execute -> ack). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3539 [FIX] Dedup PG integration test fixtures into conftest (SonarCloud) The connect-to-dev-DB + skip-if-unreachable/unmigrated block was copy-pasted across test_pg_queue_client / test_dispatch_pg / test_pg_queue_consumer (6.3% duplication on new code, over the 3% gate). Extracted into shared pg_conn / pg_client fixtures + an integration_pg_conn() helper in tests/conftest.py; the three files now use them. Behaviour unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3539 [FIX] Address Phase 9c review feedback - fairness header rebuilt from the payload on the PG run path (was dropped) — a PG-routed run now mirrors the Celery dispatch contract. - poison-message guard: surface read_ct (QueueMessage + dequeue RETURNING), drop a task that keeps failing past max_attempts (default 5, env WORKER_PG_QUEUE_CONSUMER_MAX_ATTEMPTS) with a loud ERROR carrying the payload, instead of redelivering forever. - malformed payload (missing task_name) -> distinct "missing task_name" drop log, not a misleading "unknown task None". - run() wraps poll_once() so a transient read/DB blip backs off and continues instead of tearing down the loop (the client self-recovers). - ack: WARN when delete() finds no row (task exceeded vt -> possible double-run). - __init__ validates positive tuning params; main() wires backoff_max + max_attempts via env (prefix helper); non-main-thread signal-install failure -> WARNING. - tests: poison drop, missing task_name, fairness-header propagation, multi-message batch, ack-no-row warn, construction validation, backoff growth/reset, poll-error resilience; fixed the read mock for read_ct. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3539 [FIX] Address Greptile review on the 9c consumer - [P1] batch shared-vt window: default batch_size 10 -> 1 so each message gets its own visibility window. The batch's vt is set atomically at claim but messages run sequentially, so with batch_size>1 the tail could exceed vt and be re-claimed mid-run (double-run). Batching stays opt-in; doc the vt > batch x worst-case-duration constraint. - [P2] QueueMessage.read_ct: drop the misleading =0 default (a "never claimed" state the dequeue can't produce — read_ct is always >=1). 0 would silently bypass the poison guard; now required, all callers supply it. - [P2] __init__: reject backoff_max < poll_interval (else min(poll*2, max) shrinks the backoff below poll_interval instead of growing). - [P2] dedup the ack-miss warning: client.delete() now logs at DEBUG; the consumer keeps the contextual WARNING (it names the task). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3541 [FEAT] Wire PG-queue consumer into run-worker.sh + bootstrap guard (#2047) * UN-3541 [FEAT] Wire PG-queue consumer into run-worker.sh + bootstrap guard Make the 9c PG-queue consumer (UN-3539) runnable via the normal worker flow, safely. Split from #2045 to keep that PR focused on consumer logic. - Launcher (pg_queue_consumer/__main__.py): set WORKER_TYPE to the source worker (default notification) BEFORE `import worker`, so the right tasks register. worker.py loads exactly one worker type's tasks; a bare import would load the general worker's and drop every notification as unknown. - run-worker.sh: `pg-queue-consumer` type runs `python -m pg_queue_consumer` (not a celery command) from the workers root; queue via env. - Startup guard: PgQueueConsumer.run() refuses to start on an empty task registry — fail loud instead of silently dropping every message. - Drop the hard-coded 8086 health port (consumer runs no health server). Integration fixes found during live dev-test (real send_webhook_notification end-to-end → Slack HTTP 200): - Opt-in status: consumer is not part of `all`; shown only when running. - Log path: detach writes to an absolute $worker_dir/$type.log so -L/-C find it (also fixes the same latent bug for pluggable workers). - PID discovery: get_worker_pids matches the `python -m` invocation, so --status / -k / -r work for the consumer. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3541 [FIX] Address PR #2047 review feedback - run-worker.sh: verify liveness after a detached launch (kill -0 + tail); `set -e` doesn't apply to `&`, so a fork that died on startup was reported as "started" — acute for the health-port-less consumer (High). - consumer.py: log the registered application task names at startup so a *wrong* (non-empty but mismatched) registry is diagnosable — the guard only catches an *empty* one (Medium observability). - consumer.py: type _env() with a TypeVar (was `cast: type -> object`, erasing types at the typed __init__) and name the offending var on a bad cast instead of a context-free ValueError (Medium). - tests: add the guard's positive / require_tasks=False bypass / built-in filter arms (only the failure arm was covered). - get_worker_pids: warn on pgrep rc>1 (operational/regex error) instead of collapsing it to "not running" (Low). - run-worker.sh: extract the repeated "pg_queue_consumer" literal into a readonly constant (SonarCloud S1192). - Comment accuracy: build_celery_app configures but does not import tasks; `notification` is the worker that owns the leaf task, not the task itself; generalise the "every notification dropped" example. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3541 [FIX] Address Greptile review feedback - __main__.py: move the WORKER_TYPE mutation + `import worker` bootstrap into a guarded _bootstrap_and_run() called only under `__name__ == "__main__"`, so an accidental import (test/IDE/type-checker walking __main__) no longer overwrites WORKER_TYPE or triggers a full worker-app bootstrap. - run-worker.sh: list `pg-queue-consumer` in the usage/--help worker types, plus a note for its WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE / _QUEUE overrides. - run-worker.sh: clarify the post-launch liveness check is a best-effort fast-fail for *immediate* (sub-second) crash-on-import/bad-config faults, not a connectivity check; kept general (an immediate crash can hit any worker) and noted the `all` subshells overlap the 1s with inter-launch sleep. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3544 [FEAT] PG-queue consumer liveness endpoint (poll-loop heartbeat) (#2051) * UN-3544 [FEAT] PG-queue consumer liveness endpoint (poll-loop heartbeat) Give the consumer a /health HTTP endpoint for K8s liveness probing, like every other worker — but keyed only on a poll-loop heartbeat. - consumer.py: track _last_poll_monotonic (refreshed at the top of poll_once, so a loop wedged on a long task goes stale and is detectable — which pgrep-based --status and the launch-liveness check cannot see). Expose seconds_since_last_poll() / is_poll_stale(). main() starts a LivenessServer when WORKER_PG_QUEUE_CONSUMER_HEALTH_PORT is set (opt-in), stops it on exit. - LivenessServer: tiny stdlib HTTP server (/health, /healthz, /livez) → 200 while fresh, 503 once stale. Deliberately lean: a liveness probe must report only "is this process making progress?", NOT broker/API reachability or resource pressure (those would crash-loop a healthy consumer on a blip). So it does NOT reuse the shared HealthChecker (which also bundles an api_connectivity check that is both wrong for liveness and currently broken — its `from .api_client_singleton` import points at a non-existent module; tracked separately). - run-worker.sh: default port 8090 (outside the 8080-8089 core range, no collision), exported opt-in; documented in --help. - tests: heartbeat fresh/stale + a real bind-and-GET 200->503 endpoint test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3544 [FIX] Address PR #2051 review feedback Important: - Liveness bind failure (OSError/port-in-use) now degrades gracefully: log and continue probe-less instead of aborting the consumer before it polls. Verified live (2nd consumer on a taken :8090 keeps draining). - run-worker.sh: 8090 collided with the first auto-discovered pluggable worker (8090 + count); pluggable discovery now starts at 8091, 8090 reserved. - LivenessServer.stop() is defensive (can't raise into main()'s finally and mask the real run() exception) and warns if the thread outlives the join. - Empty WORKER_PG_QUEUE_CONSUMER_HEALTH_PORT now hits the clean opt-out (_env treats "" as unset) instead of int("") crashing at launch. Suggestions: - LivenessServer: guard double start(); reset state in stop(); wrap serve_forever so a thread crash is logged; route handler errors to the logger (log_message=pass was hiding log_error too); guard wfile.write against client disconnects; single clock read per request. - Type _httpd/_thread as HTTPServer|None / Thread|None via TYPE_CHECKING (drops the Any import); restores static checking. - run-worker.sh: status line shows the effective -p override, not the map default. - Docstrings: phrase the helper trigger in terms of `port`; note 0.0.0.0 bind; document that a fast-failing loop stays healthy by design (liveness must not couple to backend reachability). - tests: heartbeat-stamped-before-read (pins top-of-poll), /healthz + /livez aliases + unknown-path 404, double-start rejection. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3544 [FIX] Address SonarCloud issues - consumer.py: use logger.exception() in the liveness-bind except block (preserves the traceback; S6679). - test: lift the walrus assignment out of the PgQueueConsumer() argument list — plain `client = MagicMock()` first (clearer; S6328). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3544 [FIX] Address Greptile review feedback - LivenessServer handler strips the query string before matching paths (self.path includes it), so /health?probe=k8s returns 200 not 404. Added a query-string case to the alias test. - Document the stale-threshold trade-off prominently: the heartbeat is frozen during task execution, so WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS is also an upper bound on single-task wall-clock (a longer task trips the probe → restart → redelivery). 60s suits the sub-second leaf; raise it above max(batch_size x worst_case_task_seconds, backoff_max) for longer tasks. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3546 [FEAT] Priority-ordered PG-queue dequeue + concurrency-safe claim (#2052) * UN-3546 [FEAT] Priority-ordered PG-queue dequeue + concurrency-safe claim Start enforcing the load-independent part of fairness — pipeline_priority (L3) — directly in the single-table dequeue: higher priority is claimed first, FIFO (msg_id) within a priority. The org-tier (L1) / workload (L2) axes + burst_max admission stay deferred to the fair-admission orchestrator. - Schema: add `priority` (smallint, default 5 = FairnessKey.DEFAULT_PRIORITY) to pg_queue_message; swap the dequeue index to (queue_name, priority DESC, msg_id) so the priority-ordered claim stays an indexed top-N. - Enqueue: dispatch() writes priority from fairness.pipeline_priority; a bare dispatch (fairness=None) writes the neutral default. - Dequeue: ORDER BY priority DESC, msg_id. Also fixes a latent concurrency bug in the original dequeue (9a): `UPDATE ... WHERE msg_id IN (SELECT ... FOR UPDATE SKIP LOCKED LIMIT n)` can OVER-CLAIM under concurrent writers — EvalPlanQual re-evaluates the LIMIT subquery when a row it tried to lock was concurrently touched, so one claim can return more than n rows. Switched to the canonical PGMQ-safe shape: lock candidates in a CTE, then `UPDATE ... FROM locked WHERE q.msg_id = locked.msg_id`, which locks exactly n rows once. The trailing SELECT re-orders RETURNING (which is otherwise unspecified) so batched claims come back in priority order too. Tests: priority selection (one-at-a-time) + batch ordering against real Postgres; send writes priority; dispatch wiring (fairness + neutral default); read param order. Verified live end-to-end via dispatch()->claim-order (9>7>3a>3b>1). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3546 [FIX] Address PR #2052 review feedback - Validate priority at the write boundary: client.send() raises ValueError on out-of-range (mirrors its vt_seconds/qty guards) — an out-of-range value would silently jump/sink the row in the priority DESC claim order. - Add a DB CheckConstraint (priority 1..10) as the backstop no ORM/raw writer can bypass (migration 0003). check= (not condition=) — repo is on Django 4.2. - Soften the "indexed top-N" comments (client.py + models.py): the dequeue is an index walk with vt<=now() as a per-row filter, NOT a guaranteed top-N — vt is not in the index, so in-flight (future-vt) high-priority rows are scanned past on each claim; the orchestrator's admission is the high-backlog answer. Update the module docstring to the CTE FROM-join shape; fix the fairness.DEFAULT_PRIORITY symbol reference; drop the duplicated param-order comment. - Tests: send() range-guard (parametrized) + DB CheckConstraint backstop; concurrent-writer over-claim guard (two readers, no batch exceeds qty — the regression test for the EvalPlanQual fix); vt × priority (visible low beats invisible high); FIFO-within-band for multi-member batch bands. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3546 [FIX] Address Greptile review feedback - Concurrency test: assert the drain worker terminated after join (a hung worker now fails the test instead of passing silently while conn_b.close() races its in-flight queries). - Priority-bounds drift guard: backend models.py and workers fairness.py are separate codebases that can't import each other, so the DB constraint bounds (1/10) duplicate fairness.MIN/MAX_PRIORITY. Replaced the hardcoded "42" reject test with test_db_check_constraint_matches_fairness_bounds — raw-inserts at MIN/MAX (accepted) and MIN-1/MAX+1 (CheckViolation), pinning the DB constraint to the fairness range so a future widening that misses one side fails loudly. Documented the canonical source in the model comment. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3546 [DOCS] Note check=/condition= handling on the constraint migration Breadcrumb for a future Django upgrade: `check=` is correct on the pinned Django 4.2 (deprecated 5.1, removed 6.0); fresh installs always replay under the shipped Django, so leave it. When the pin reaches >= 6.0, squash (or do the behaviour-preserving check= -> condition= edit) so a from-scratch migrate runs. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3548 [FEAT] PgBarrier — Postgres fan-in barrier (3rd WORKER_BARRIER_BACKEND) (#2053) * UN-3548 [FEAT] PgBarrier — Postgres fan-in barrier (3rd WORKER_BARRIER_BACKEND) Add a Postgres Barrier substrate selected by WORKER_BARRIER_BACKEND=pg (default stays chord). Moves the fan-in aggregation ("wait for N header tasks, then fire the callback with their results") onto a pg_barrier_state row — the same DB that holds the PG queue, so an execution can coordinate without Redis/RabbitMQ. The 9e pipeline on-ramp primitive. Mirrors RedisDecrBarrier 1:1 — same Barrier protocol, fairness plumbing, Celery-dispatched header tasks with .link/.link_error, empty->None, missing-execution_id->raise, mid-loop dispatch cleanup. Defaults-off, zero behaviour change until the flag flips. - Schema (backend/pg_queue): pg_barrier_state (execution_id PK, remaining, results jsonb, aborted, expires_at) + migration 0004. - pg_barrier.py: PgBarrier + barrier_pg_decr_and_check / barrier_pg_abort. Atomic decrement is ONE statement (UPDATE ... SET remaining = remaining-1, results = results || jsonb_build_array(%s) ... RETURNING remaining, results, aborted) — row lock serialises concurrent decrements so exactly one sees 0; no Lua. Guards: reads aborted in the same statement (never fires partial), row-missing / negative-remaining clean up without firing, callback dispatched BEFORE row delete. Orphan bound via expires_at + opportunistic sweep in enqueue (periodic sweep is the backstop). - __init__.py: BarrierBackend.PG -> PgBarrier() in get_barrier(). Tests: protocol shape, TTL env validation, enqueue (upsert/links/fairness/ stale-reset/expiry-sweep/mid-loop-cleanup), decr paths (pending/complete-fires/ aborted/negative/missing/unserialisable), abort (claim+delete/dedup), and a real two-connection decrement-atomicity check (exactly one sees 0). Selector PG case. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3548 [FIX] Address PR #2053 review feedback High: - Abort is now ONE atomic statement: `WITH claimed AS (DELETE ... RETURNING) ...` — claim+teardown in a single transaction (no claimed-but-not-deleted window; a crash rolls back so a sibling retries). This makes the `aborted` column redundant — dropped it; the decrement's "row missing -> abandoned" branch now covers the failed-task case. The callback can only fire when remaining hits 0 (all tasks succeeded), so a failed task (which deletes the row) can never let a partial-results fire. - Dropped the per-enqueue global orphan sweep (unbounded DELETE on the hot path, deadlock-prone, shared the UPSERT txn). Reclaim is a future periodic sweep. - A NUL byte survives json.dumps but jsonb rejects it -> catch the DataError and tear the barrier down (fail fast) instead of hanging to expiry. Medium: - Post-dispatch row delete is best-effort (logged, not raised) so a delete error can't mask the already-fired callback; documented the no-double-fire invariant (last decrement + max_retries=0). - Added a DB CheckConstraint (expires_at > created_at) — the one writer-proof invariant; Meta comment warns off a `remaining >= 0` check (teardown needs negative). Softened the "periodic sweep" comments to future/not-yet-shipped. Low: - Extracted shared `barrier_ttl_seconds()` + `CallbackDescriptor` into barrier.py; both backends import them (redis keeps back-compat aliases). signature_kwargs dict instead of inline ** spread. Atomicity comment notes the per-transaction premise. Tests: callback-dispatch-failure-preserves-row; decrement-after-abort-no-fire; atomicity through barrier_pg_decr_and_check (two threads, exactly one fires); list-result-as-single-element; NUL-byte teardown; DB-constraint; max_retries=0. 92 barrier tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3548 [DOCS] Drop stale aborted-column reference in PgBarrier docstring The wire-model docstring's enqueue step still listed `aborted = false` as an UPSERT column after the column was removed (abort now dedups via DELETE … RETURNING / row existence). Remove it so a reader doesn't hunt for — or re-add — a column that no longer exists. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * UN-3553 [FEAT] PG Queue 9d slice 1 — leader-election lease (orchestrator_lock) (#2056) * UN-3553 [FEAT] PG Queue 9d slice 1 — leader-election lease (orchestrator_lock) First slice of 9d (orchestrator/reaper): the singleton-guarantee primitive the reaper loop and a future fair-admission gate hang off. Ships dark — nothing acquires leadership yet, so merging changes no runtime behaviour. Why now: 9d was skipped in the merged spine (9c -> liveness -> priority -> PgBarrier) and is the safety net 9e needs — without a reaper, every at-least-once hang / orphaned barrier bottoms out at the 6h TTL with no recovery. The reaper must run as exactly one instance, so leader election is the foundation. Lease, not advisory lock: leadership is a TTL'd row UPDATE (take it if the leader is free or its lease is stale), not pg_advisory_lock. Session-scoped advisory locks don't survive the transaction-pooled PgBouncer the queue connects through (UN-3533) — a plain UPDATE is one transaction, pooling-safe. All time comparisons are server-side (now()), so candidate clock skew can't split leadership. - backend/pg_queue: PgOrchestratorLock single-row model (id PK, leader, acquired_at) + CheckConstraint(id=1); generated migration 0005 + a reversible RunPython seeding the one free row. Free = empty leader (follows the PgQueueMessage.org_id no-nullable-text convention). - workers/queue_backend/pg_queue/leader_election.py: LeaderLease (try_acquire/renew/release), lease_seconds_from_env() (default 10s, loud-on-misconfig), default_worker_id(). Instance-owned self-recovering connection. - tests: 20 real-PG tests. Load-bearing properties — concurrent try_acquire yields exactly one winner; renew returns False after a stale-lease takeover (the signal that stops a stalled leader). Plus lease-expiry takeover, release-frees-immediately, non-holder no-ops, env validation, single-row constraint. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3553 [FEAT] Address review: connection ownership, logging, recovery tests Toolkit + SonarCloud review on #2056: - [High] _owns_conn ownership guard: LeaderLease now mirrors PgQueueClient — an injected connection is never closed/swapped on a transient error (it would otherwise silently re-point an injected TEST_DB_/caller connection at a fresh DB_-env one). _get_conn only recreates an OWNED missing/closed conn. - [Medium] Log the owned-connection discard in _cursor (worker_id + exc type) — a silent rebuild on the reaper singleton correlates with missed renews. - [Medium] Test the recovery machinery: owned-conn recovered on OperationalError, owned-conn recovered when rollback fails, injected-conn never swapped. Plus two documented-invariant gaps — same-holder re-acquire on a fresh lease returns False, and release after a takeover is a no-op. - [Low] release() branches on rowcount — only logs "released" when it really freed the lease; a no-op release logs debug (truthful post-mortems). - [Low] Scope the lease_seconds<=0 guard to the explicit-arg branch (dead on the env path, which already rejects <=0). - [Low] Document the exception-propagation contract (raise == "leadership unknown, stop acting"), relabel the durable Usage example. - [Low] Migration: note the seed row is load-bearing (future reaper-bootstrap should self-heal with INSERT ... ON CONFLICT DO NOTHING). - SonarCloud S117: rename the migration's get_model local to lock_model. 25 leader-election tests pass; makemigrations --check clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3553 [FEAT] Address Greptile: idempotent worker id + clearer renew log - default_worker_id() is now cached (functools.cache) → idempotent per process, so a caller passing it inline in a retry/restart loop can't drift the worker id out from under renew()/release(). Lazy (first-call), so it's fixed after a fork rather than shared across children. Test asserts idempotency. - renew()'s failure warning now reads "not the current leader (taken over by another candidate, or the lease was never held)" — accurate for the non-holder-renew case too, and fixes the "took over"->"taken over" grammar. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3554 [FEAT] PG Queue 9d slice 2 — reaper process + barrier-orphan sweep (#2058) * UN-3554 [FEAT] PG Queue 9d slice 2 — reaper process + barrier-orphan sweep Builds on the leader-election lease (UN-3553): stands up the reaper process — the leader-elected recovery loop — with its first recovery job, the barrier-orphan sweep. Ships dark (launched explicitly, never in the default worker set). - reaper.py: - sweep_expired_barriers(conn): DELETE pg_barrier_state WHERE expires_at < now() RETURNING + loud per-orphan WARNING. The documented PgBarrier backstop — reclaims barriers whose header tasks never all completed; a late in-flight decrement then finds no row and abandons (existing semantics). Execution terminal-status recovery is 9e's job. - PgReaper: leader-elected loop. Each cycle renews (steps down to standby if renew() returns False), else tries to acquire; sweeps ONLY while leader. run() loops with graceful SIGTERM/SIGINT shutdown + lease release on exit. Guard: cycle interval must be shorter than the lease window, or the leader thrashes leadership between renews. - reaper_interval_from_env() (WORKER_PG_REAPER_INTERVAL_SECONDS, default 5s), main(), python -m queue_backend.pg_queue.reaper entrypoint. - leader_election.py: expose lease_seconds property (reaper validates its cycle against it). - test_pg_reaper.py: 15 tests — env/construction guards (interval < lease), leadership gating (sweeps only when leader, steps down on renew-fail, releases on stop), real-PG sweep (reclaims only expired, leaves fresh). Out of scope: run-worker.sh wiring + liveness (followup, like 9c-followup); pipeline recovery (counter reconstruction, per-stage re-enqueue) deferred to 9e where there's a real PG pipeline to test against. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3554 [FEAT] Address review: sweep rollback, conn recovery, tick contract Toolkit + SonarCloud review on #2058: - [CRITICAL] sweep_expired_barriers now rolls back on error before re-raising (conn is manual-commit; an un-rolled-back failure left it in an aborted-txn state, poisoning every later cycle → silent self-perpetuating stall). This also clears SonarCloud's C-reliability gate. - [HIGH] On a failed sweep PgReaper discards its OWNED connection so the next tick reconnects — covers a poisoned/dead handle that `.closed` alone misses. - [MEDIUM] renew() raising now sets _is_leader=False before propagating (honours the lease's "raise == stop acting" contract). - [MEDIUM] release() failure on shutdown is logged (with the lease-window note) instead of silently suppressed. - [MEDIUM] signal-handler ValueError is re-raised unless we're off the main thread (don't mislabel an unrelated ValueError). - [MEDIUM/type-design] tick() returns a TickOutcome(was_leader, reclaimed) NamedTuple instead of an overloaded `-1` int sentinel; added an is_leader property; lease param typed against a new LeaderLeaseLike Protocol. - [LOW] Reworded the sweep race comment, the step-down log (same-cycle re-acquire), and the run() self-recovery comment for accuracy. - Tests: +8 — run() swallows a tick error; owned-conn recreated-when-closed; injected-conn never swapped; failed-sweep discards owned conn; sweep SQL contract (no DB); sweep rolls back on error; step-down-then-reacquire; renew-raising steps down. 23 total; drive paths via is_leader, no private-flag poking. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3554 [FIX] Use pytest.approx for float-equality asserts (SonarCloud S1244) The two reaper-interval asserts compared float returns with ==; the values are exactly representable so it was harmless, but pytest.approx is the correct idiom and clears the S1244 reliability bugs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3554 [FEAT] Address Greptile: manual-commit Layer-4 fixture, close owned conn - The real-Postgres fixture (barrier_conn) was autocommit, which made sweep_expired_barriers' own commit() a no-op and its rollback unreachable — so Layer 4 tested a different mode than the production reaper (create_pg_connection is manual-commit). Switched the fixture to manual-commit and added explicit commits to the seed/read/cleanup helpers, so the real-DB tests now exercise the sweep's commit (and rollback) in production mode. - run() now closes its OWNED sweep connection on shutdown (an injected one is the caller's). Harmless for the main() process but keeps PgReaper clean if ever embedded / test-driven. 23 tests pass; ruff clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3555 [FEAT] PG Queue 9d slice-2 followup — run-worker.sh reaper type + pg-queue set (#2059) * UN-3555 [FEAT] PG Queue 9d slice-2 followup — run-worker.sh reaper type + pg-queue set Wires the reaper (UN-3554) into run-worker.sh and adds a pg-queue set so the whole PG-queue group launches in one shot. Mirrors the 9c -> 9c-followup split (launcher wiring as its own slice). Liveness probe is a separate follow-on. - workers/pg_queue_reaper/: thin entrypoint package (python -m pg_queue_reaper -> queue_backend.pg_queue.reaper.main). No worker-app bootstrap (the reaper runs no Celery tasks), unlike pg_queue_consumer; exists so the process has a stable name run-worker.sh can launch + pgrep-match. - run-worker.sh: - reaper / pg-queue-reaper type — opt-in (NOT in `all`), launches `python -m pg_queue_reaper`, runs from workers root, --status/-k/-r match via the `-m` pgrep branch (now covers consumer + reaper). Lease/interval env documented in --help. - pg / pg-queue SET — run_pg_queue_set launches consumer + reaper together (always detached, like `all`). Restart (-r pg-queue) kills both members then relaunches. list_core_worker_dirs skips the set alias (no phantom status entry). Help documents the Celery `all` set and the PG `pg-queue` set as independent, runnable in parallel for a dual-transport (strangler-fig) setup. Dev-tested live: `run-worker.sh reaper -d` acquires leadership + ticks and shows RUNNING in --status; `run-worker.sh pg-queue` brings up consumer + reaper, both RUNNING, reaper leader, no phantom set entry; bash -n clean; --help renders. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3555 [FEAT] Address review: set start-failure propagation, restart-kill guard Toolkit review on #2059: - [High] run_pg_queue_set swallowed member start-failures (backgrounded subshells' status was lost, banner+return 0 unconditional). Now each member runs in a FOREGROUND subshell so run_worker's own `return 1` on a crash-on-start is captured; the set returns non-zero if any member fails. (The reviewer's kill -0 on `$!` would false-fail — that PID is the launcher subshell, which exits the instant it backgrounds the nohup'd worker; the foreground-subshell return value is the correct signal and isolates `cd`.) Documented that the set always runs detached (ignores -d). - [Medium] Dispatch now `|| exit 1` so a member start-failure reaches the script exit code — the only programmatic startup signal (reaper has no health port yet). - [Medium] Set-restart aggregates kill_one_worker failures and aborts the relaunch if a member survives SIGKILL (avoids a duplicate consumer double-polling Postgres). Mirrors kill_workers' discipline. - [Medium/minor] Startup banner prints `Queues: n/a` when empty (reaper). - [Low] Reworded pg_queue_reaper/__main__ docstring: the launcher DOES export WORKER_TYPE for every worker; the accurate claim is the reaper neither reads nor mutates it (vs the consumer overwriting it before `import worker`). - [Low] Added a smoke test that pg_queue_reaper.__main__ re-exports the real reaper main (guards the `python -m pg_queue_reaper` launch path against an ImportError regression). Dev-tested: `run-worker.sh pg-queue` returns exit 0 with both members up; 24 reaper tests pass; bash -n + ruff clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3555 [FEAT] Address Greptile: set partial-start teardown + --logs set alias - run_pg_queue_set: on a partial start-failure (one member up, the other crashed) tear the whole set down before returning 1 — kill both members so a restart-on-failure relaunch can't spawn a second instance over the survivor (the consumer would double-poll Postgres). All-or-nothing, mirroring the restart path's discipline. - tail_logs: handle the pg/pg-queue set alias — `--logs pg-queue` now tails both member logs (pg_queue_consumer + pg_queue_reaper) instead of looking for a non-existent workers/pg-queue/pg-queue.log and printing a misleading "no log file" error. Mirrors how list_core_worker_dirs skips the set value. Dev-tested: `--logs pg-queue` tails 2 files (consumer + reaper); bash -n clean; 24 reaper tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3556 [FEAT] PG Queue 9d — reaper liveness probe (heartbeat + is_leader) (#2061) * UN-3556 [FEAT] PG Queue 9d — reaper liveness probe (heartbeat + is_leader) Closes the gap flagged in #2059's review: a reaper that crashes after startup was invisible (opt-in skip in --status + no health port). Mirrors the consumer's liveness (UN-3544). - PgReaper heartbeat: _last_tick_monotonic stamped at the START of every tick (a standby tick counts as progress — liveness tracks the loop, not leadership) + seconds_since_last_tick() / is_tick_stale(). - ReaperLivenessServer: lean HTTP probe (mirrors the consumer's LivenessServer) — /health (also /healthz, /livez) returns 200 while the tick loop is fresh, 503 when stale. Payload also surfaces is_leader (which pod holds the lease — useful for 9e debugging). The 200/503 verdict is PURELY the heartbeat, never leadership (a standby is healthy) or DB reachability (a blip must not crash-loop a fine process). - main() wires it from WORKER_PG_REAPER_HEALTH_PORT (unset → no server, no stray port); staleness window from WORKER_PG_REAPER_HEALTH_STALE_SECONDS (default 30s, comfortably above the 5s tick interval). Bind failure degrades gracefully (logs, runs probe-less). - run-worker.sh: reserve port 8086 for the reaper, export the health-port env in the reaper special-case, document the two new env vars in --help. - Tests (+13, 37 total): heartbeat fresh/stale + tick refresh; liveness server 200-fresh / 503-stale / is_leader reflected / 404 / double-start; health staleness env default+override+invalid; server-disabled-when-no-port. Dev-tested live: `run-worker.sh reaper -d` → GET :8086/health → {"status":"healthy","check":"pg_reaper_tick","is_leader":true,...}. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3556 [REFACTOR] Extract shared LivenessServer (SonarCloud duplication) SonarCloud flagged the new ReaperLivenessServer as duplicating the consumer's LivenessServer (~19 lines of HTTP-probe boilerplate, over the 3% new-code gate). Extracted one generic LivenessServer into queue_backend/pg_queue/liveness.py — parameterised by a freshness callable + the payload's check/age labels + an optional extra-status callable (the reaper's is_leader). Both sides are now thin subclasses that preserve their exact constructor signatures and wire payloads: - consumer LivenessServer(consumer, port=, stale_after=) → check="pg_queue_poll", seconds_since_last_poll (unchanged on the wire; its tests pass untouched). - ReaperLivenessServer(reaper, port=, stale_after=) → check="pg_reaper_tick", seconds_since_last_tick, is_leader. The boilerplate now lives once → duplication cleared, consumer behaviour preserved. reaper 37 tests + consumer liveness/health tests green; ruff clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3556 [FEAT] Address review: validate health port, type/guard liveness, tests Toolkit review (10 findings; several already resolved by the dedup refactor 0d2b3f721 — the threading-alias, the query-strip comment, and the extract-shared-server follow-up itself): - [Medium] Port parse: extracted _reaper_health_port_from_env() — names the var on a bad value (no more context-free int('abc') crash) and range-checks 0-65535 at parse time, so an out-of-range value can't escape the bind catch as OverflowError inside start(). main() uses it. - [Medium] liveness.py: typed _httpd/_thread as HTTPServer|None / Thread|None via TYPE_CHECKING (was Any in the shared server) — restores the lifecycle invariant + type-checking on .shutdown()/.join()/etc. - [Low] LivenessServer.__init__ re-validates stale_after > 0 (a direct caller could otherwise build an always-503 probe). - [Low] bound_port docstring: clarified the port=0 / not-started case. - [Low] _DEFAULT_HEALTH_STALE_SECONDS comment references the interval constant, not a hard-coded "5s". - [Medium/Low test gaps] +9 tests: port-env helper (unset/empty/valid/non-int named/out-of-range); main() wiring (parsed port reaches the wiring + health stopped in finally; port=None when unset); _maybe_start_health_server OSError graceful-degrade → None + logger.exception; stale_after<=0 constructor guard. reaper 46 + consumer liveness 10 green; ruff clean. The shared-server extraction (flagged as a follow-up) was already done in 0d2b3f721. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3556 [FEAT] Address Greptile: protect core payload + per-process log label - extra_status_fn could silently clobber core payload fields (status / check / age_key / stale_after_seconds) that a monitor reads. Now the handler builds extra fields first and overlays the core fields, so core ALWAYS wins — a future caller's extra dict can't corrupt the status a monitor parses. Test: an extra_status_fn returning {"status":"HACKED",...} leaves status "healthy" and check intact, while a non-reserved extra key is preserved. - The dedup refactor moved the consumer's liveness warnings to the shared liveness logger with generic text, so log-based filtering keyed on the old "PG-queue consumer: ..." would miss them. Added a log_label param (default "pg-queue"); consumer passes "pg-queue consumer", reaper "pg-queue reaper", so the messages stay attributable to the source process. 57 tests green (reaper 47 + consumer liveness 10); ruff clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3560 [FEAT] PG Queue 9 — run-worker.sh `-L celery` log alias (#2066) * UN-3560 [FEAT] PG Queue 9 — run-worker.sh `-L celery` log alias `./run-worker.sh -L pg-queue` already tails the PG-queue set's logs, but there was no symmetric way to tail only the Celery set: `-L` (no arg) tails EVERYTHING (Celery + PG-queue consumer/reaper), since list_core_worker_dirs includes the PG worker dirs. Add a `-L celery` log alias (mirror of `-L pg-queue`): tails every worker log EXCEPT the PG-queue members (pg_queue_consumer, pg_queue_reaper). Logs-only — the Celery set is still run via 'all'. run-worker.sh only: CELERY_SET constant + a branch in tail_logs() + usage/examples. Dev-tested with stub PG logs: -L = 11 files, -L celery = 9 (PG excluded), -L pg-queue = 2 (PG only). bash -n clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3560 [FIX] -L celery review — complement wording + single-source PG members + dedup Address PR #2066 toolkit review: - Comment accuracy (P1, x2): reword 'mirrors PG_QUEUE_SET' / 'mirror of the pg-queue alias' — the celery set is the COMPLEMENT (all minus the two PG members), in a different branch, not a mirror. Drop the directional 'below'. - Modeling (P2): add a single 'PG_QUEUE_MEMBERS' source of truth (readonly assoc array) so 'celery = all - pg-queue' stays correct by construction; the celery branch tests membership via it instead of hand-rolling consumer/reaper. - Simplification (P1): collapse the near-duplicate 'all' and 'celery' tail_logs branches into one loop with a membership-guarded skip. Deferred (P2, inherited): set-aware empty-result message for the shared zero-log guard — spans all three set aliases, best done as one follow-up. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3559 [FEAT] PG Queue 9e PR 1 — execution transport seam (inert) (#2062) * UN-3559 [FEAT] PG Queue 9e PR 1 — execution transport seam (inert) Establish the per-execution transport seam for the 9e coupled-pipeline migration: the transport a workflow execution rides ("celery" | "pg_queue") is resolved once at the creation chokepoint and carried in the task payload. Inert — transport always resolves to "celery", so behaviour is unchanged. Design (chosen = payload-carry, not a WorkflowExecution column): workers/queue_backend/pg_queue/9e-design.md. - core: WorkflowTransport enum + DEFAULT_WORKFLOW_TRANSPORT (shared vocabulary). - backend: resolve_transport() hardwired to celery (signature shaped for PR 3's Flipt wiring); create-execution internal API returns "transport"; execute_workflow_async adds it to the async_execute_bin kwargs. - workers: scheduler threads transport from the create response into the dispatch kwargs; async_execute_bin_general / _execute_general_workflow carry it onto the live WorkflowContextData. - tests: backend resolver + enum (5), worker WorkflowContextData carry/default (2), dispatch characterisation updated (+ backend-resolved-transport thread). Out of scope: live PG routing + per-batch idempotency key (PR 2); Flipt canary wiring (PR 3); rollout (ops). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3559 [FIX] 9e PR 1 review — fail-closed transport coercion + doc/test fixes Address PR #2062 review (PR Review Toolkit findings): - Validation/silent-failure: add normalize_transport() (core) — fail-closed coercion of any inbound transport to a known value (unknown/None -> celery + warn). Applied at the scheduler read boundary and in WorkflowContextData __post_init__, so a garbage payload value can't reach the PR 2 fan-out read. - Comment accuracy: WorkflowContextData.transport comment now present-tense (carried in PR 1; fan-out read lands in PR 2). - Dead-code hygiene: add 'transport' to EXECUTION_EXCLUDED_PARAMS so the legacy execute_bin -> create_workflow_execution path can't TypeError. - Two-resolution-sites: documented the deliberate two-site design + the PR 3 single-chokepoint requirement at execute_workflow_async. - transport.py: drop the unused logger; leave a PR-3 fail-closed marker. - Design doc: correct anchors (transport.py / internal_api_views / workflow_ helper, not execution.py:126), class name (WorkflowContextData not WorkflowExecutionContext), scope claims (stage-1 only in PR 1), and remove the hard-coded Flipt version/date/live-state. - Tests: normalize_transport (passthrough / invalid->celery / None / logging), WorkflowContextData invalid-transport coercion. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3559 [FIX] 9e PR 1 — resolve transport after execution_id guard (greptile P1) Move the normalize_transport(...) extraction below the `if not execution_id` guard in _execute_scheduled_workflow. Previously it ran before the guard, so an error response with no execution_id logged a misleading `[exec:None]` context and discarded the computed transport. Now transport is only resolved once execution_id is known non-empty. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3559 [FIX] 9e PR 1 — SonarCloud: drop unused resolve_transport params + dict literal Address SonarCloud code smells on PR #2062: - python:S1172 (x3): resolve_transport() no longer declares the unused workflow_id/pipeline_id/organization_id params — the PR-1 seam is inert and needs no inputs. PR 3 reintroduces them (keyed for Flipt) when it wires the evaluation; the two call sites (internal_api_views view, execute_workflow_async) now call resolve_transport() with no args. Tests updated. - Replace dict(...) constructor with a {...} literal in test_workflow_context_transport._make_context. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3561 [FEAT] PG Queue 9e PR 2a — live-PG-pipeline inert foundation (dispatch backend override + barrier decrement core) (#2067) * UN-3561 [FEAT] PG Queue 9e PR 2a — live-PG-pipeline inert foundation (dispatch backend override + barrier decrement core) First slice of 9e PR 2 (the live PG execution pipeline), split 2a/2b/2c. 2a is the inert foundation: the two seams the live switch (2c) consumes, each with zero behaviour change on the default path. - dispatch(backend=...): per-call transport override. None (default, every call site today) keeps the env allow-list decision via select_backend — byte-identical. When set it wins over the allow-list, so 2c can route a whole execution's header/callback onto PG without opting their task names into WORKER_PG_QUEUE_ENABLED_TASKS (allow-list is for leaf tasks; the pipeline's migration unit is the execution). - Extract _barrier_pg_decrement(...) plain core out of the @worker_task barrier_pg_decr_and_check (now a thin delegator). 2c calls the core in-body on the PG-consumed path (a PG-consumed task fires no Celery .link, so the decrement runs in-body — fire-and-forget self-chaining). Inert by construction: default barrier backend is chord (Celery executions never import the PgBarrier module), and no call site passes backend=. Net behaviour change: NONE. Transport threading + live switch land in 2c; per-batch idempotency in 2b. Tests: dispatch backend-override (3) + decrement-core extraction (3, incl. real-PG in-body decrement + verbatim delegation). pg_barrier/dispatch/ barrier/routing suites green; ruff clean; worker-app bootstrap clean under WORKER_BARRIER_BACKEND=pg (both barrier tasks registered, get_barrier→PgBarrier). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3561 address review (muhammad-ali-e): loud in-transaction guard, resolve_backend helper, comment/test fixes - pg_barrier: enforce the "own committed transaction" decrement contract loudly — _barrier_pg_decrement raises at entry if the shared connection is mid-transaction (was prose-only; the 2c in-body caller is the real risk). Safe for existing paths: Celery .link enters idle, tests use autocommit conns → always idle. Fix the inaccurate "one call = one _cursor() txn" parenthetical (the delete paths open a 2nd txn) and drop the rot-prone "(fire-and-forget self-chaining)" jargon. - routing: extract resolve_backend(task_name, override) — the override-wins- else-allow-list precedence now lives in one self-documenting place (2c reuses it); dispatch() calls it instead of inlining the None-means-auto rule. - dispatch: reword the backend= docstring — avoid the "payload" term collision (local to_payload var) and the stale routing.py cross-ref; point at the live carrier WorkflowContextData.transport + 9e-design.md. - tests: +fairness-reaches-row on the backend= override path; +real-row wrapper test that catches core-param drift (the mocked delegation test couldn't); +open-transaction guard test. Kept the mocked test as the keyword-forwarding pin. Deferred (LOW, reviewer-aligned): BarrierDecrementResult TypedDict union — lands in 2c when the in-body caller actually branches on the status. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3561 address greptile (3× P2): docstring refs + routing-log suppression note All doc/comment-only, no logic change. greptile gave 4/5 "safe to merge"; these are the staleness its findings flagged, introduced by the resolve_backend extraction in the prior review round: - dispatch module docstring + routing "Scaffold posture" now name resolve_backend (wrapping select_backend) as the routing seam, not select_backend alone. - Note at the _pg_routing_logged log-once site that it's keyed on task name only, so an override-then-allow-list cutover won't re-announce (benign: override = pipeline headers vs allow-list = leaf tasks, no overlap expected; the allow-list config is still announced by _log_allow_list_once). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * UN-3562 [FEAT] PG Queue 9e PR 2b — per-batch idempotency primitive (pg_batch_dedup + claim_batch / clear_execution_batches) (#2068) * UN-3562 [FEAT] PG Queue 9e PR 2b — per-batch idempotency primitive (pg_batch_dedup + claim_batch / clear_execution_batches) Second slice of 9e PR 2, after 2a (#2067). Inert idempotency primitive: the durable per-batch dedup marker 2c wires into the at-least-once PG path. Why: the PG queue is at-least-once, so process_file_batch can be redelivered after a crash-before-ack → re-run batch + double-decrement the barrier (non-idempotent, max_retries=0). Recon showed existing per-file protection is only partial (Redis lock released after write; WFE COMPLETED skips tool re-exec but not necessarily the destination write; FileHistory is cross-execution only), so a durable per-batch gate is needed; per-file status stays the partial-crash backstop. - backend: PgBatchDedup model + migration 0006 — table pg_batch_dedup with a UniqueConstraint(execution_id, batch_index) (the ON CONFLICT target; its execution_id-leading index also serves the cleanup DELETE). Django-managed, extension-free, same posture as the sibling pg_queue models. - workers (pg_barrier.py, reusing the barrier's _cursor() → one PG conn per worker child): claim_batch(execution_id, batch_index) -> bool (atomic INSERT ... ON CONFLICT DO NOTHING RETURNING; True=first/decrement, False=redelivery/skip) + clear_execution_batches(execution_id) -> int (barrier-teardown cleanup; reaper sweep is the backstop). No call-site wiring — claim/clear + batch_index threading + transport switch land in 2c. Inert: new table + two helpers, no callers. Net behaviour: NONE. Tests: +8 real-PG (first-claim, redelivery-rejected, distinct-batch, distinct-exec, concurrent-exactly-one-winner, clear-only-target, clear-empty-zero, reclaim-after-clear). Migration applied to dev DB, makemigrations --check clean, bootstrap clean under WORKER_BARRIER_BACKEND=pg. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3562 address review (muhammad-ali-e P1-P8): fix reaper-backstop docstring claim, batch_index constraint, race test, nits - P1 (verified bug): the docstrings claimed the reaper's barrier-orphan sweep reclaims orphaned pg_batch_dedup markers — it doesn't. sweep_expired_barriers DELETEs only pg_barrier_state (no cascade), so orphaned markers leak today. Reworded both PgBatchDedup + clear_execution_batches docstrings to state the leak honestly + flag the dedup-orphan sweep as intended future work. - P4: add CheckConstraint(batch_index >= 0) (writer-proof, mirrors PgQueueMessage.priority) + a test that the DB rejects a negative index. Regenerated migration 0006 to include it. - P6: claim_batch docstring no longer says it decrements; defers the single decrement to the caller (the function only inserts the marker). - P7: generalized "partial per-file protection" → "not fully idempotent on redelivery" so the rationale can't rot. - P5: documented created_at as observability-only (future age-based sweep). - P2: REQUIRE_PG_TESTS env → skip becomes fail, so the idempotency primitive can't ship untested-green in CI where PG is expected. - P3: strengthened the race test — pre-build N=8 conns in the main thread, align claims with threading.Barrier, loop 5 trials (forces the contended ON CONFLICT path instead of a serial fast-path). - P8: hoisted import os to module top. 35 pg_barrier + 9 dedup tests green; migration applied to dev DB, makemigrations --check clean; ruff clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by:…

muhammad-ali-e commented Jun 12, 2026

View reviewed changes

muhammad-ali-e marked this pull request as ready for review June 12, 2026 14:14

greptile-apps Bot reviewed Jun 12, 2026

View reviewed changes

Comment thread workers/tests/test_pg_queue_client.py

Comment thread backend/pg_queue/models.py

muhammad-ali-e merged commit 0955df9 into feat/UN-3445-pg-queue-integration Jun 12, 2026
4 checks passed

muhammad-ali-e deleted the UN-3546-pg-queue-priority-dequeue branch June 12, 2026 14:38

Uh oh!

Conversation

muhammad-ali-e commented Jun 12, 2026

⚠️ Also fixes a latent concurrency bug in the core dequeue (9a)

What

Deliberately deferred (need load / belong with the orchestrator)

Testing

Uh oh!

coderabbitai Bot commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review skipped

Uh oh!

muhammad-ali-e left a comment

Choose a reason for hiding this comment

PR Review Toolkit — automated multi-agent review

🔴 Critical / Important

🟡 Test gaps (inline on test_pg_queue_client.py:283)

🟢 Comment / cleanup nits (inline)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

muhammad-ali-e commented Jun 12, 2026

Review feedback addressed — 15d1d45cf

Uh oh!

greptile-apps Bot commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Greptile Summary

Confidence Score: 5/5

Important Files Changed

Sequence Diagram

Uh oh!

Uh oh!

Uh oh!

muhammad-ali-e commented Jun 12, 2026

Greptile feedback addressed — b68ec9fb4

Uh oh!

Uh oh!

sonarqubecloud Bot commented Jun 12, 2026

Quality Gate passed

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

coderabbitai Bot commented Jun 12, 2026 •

edited

Loading

🟡 Test gaps (inline on `test_pg_queue_client.py:283`)

Review feedback addressed — `15d1d45cf`

greptile-apps Bot commented Jun 12, 2026 •

edited

Loading

Greptile feedback addressed — `b68ec9fb4`