UN-3541 [FEAT] Wire PG-queue consumer into run-worker.sh + bootstrap guard by muhammad-ali-e · Pull Request #2047 · Zipstack/unstract

muhammad-ali-e · 2026-06-12T09:08:12Z

Makes the 9c PG-queue consumer (UN-3539, #2045) runnable through the normal worker flow — and safe to launch. Split out so #2045 stays focused on consumer logic.

Targets the long-lived feat/UN-3445-pg-queue-integration branch (not main).

Why

The consumer is task-name routed: it reads a task_name from each PG message and runs it from the worker app's task registry. So the launch path must (a) bootstrap the right task set and (b) fail loudly if it didn't — otherwise every message is silently dropped as "unknown task".

What

Bootstrapping launcher — workers/pg_queue_consumer/__main__.py

Sets WORKER_TYPE to the source worker (default notification) before import worker. The root worker.py loads exactly one worker type's tasks.py; a bare import would load the general worker's tasks and drop every notification. Overridable via WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE.

run-worker.sh wiring

New pg-queue-consumer worker type → uv run python -m pg_queue_consumer (a plain Postgres poll loop, not a Celery/RabbitMQ worker), run from the workers root; queue routed via WORKER_PG_QUEUE_CONSUMER_QUEUE.
Opt-in: not part of ./run-worker.sh all (default behaviour unchanged — the new arch is experimental/flag-gated). Status shows it only when running, so a deliberate non-start isn't reported as STOPPED.

Startup guard — PgQueueConsumer.run(require_tasks=True)

Refuses to start on an empty task registry (a strong signal the worker app wasn't bootstrapped). Fails loud instead of silently destroying data.

Health-port fix

Dropped the hard-coded 8086 — the consumer runs no health server, so it binds no port (avoids a collision). A liveness endpoint, if added later, reads an env port.

Integration fixes found during live dev-test

Log path — detached workers now write to an absolute $worker_dir/$type.log, so -L/-C find it (also fixes the same latent bug for pluggable workers, which likewise run from the workers root).
PID discovery — get_worker_pids matches the python -m invocation, so --status / -k / -r work for the consumer.

Testing

Live end-to-end: ./run-worker.sh pg-queue-consumer drained real send_webhook_notification messages from the PG queue → Slack HTTP 200, tasks succeeded, rows deleted.
Unit: new test asserts the startup guard refuses an empty-registry app; full consumer suite green.
Verified --status (RUNNING/hidden-when-stopped), single-worker -r, and log resolution.

Out of scope (rollout)

Real liveness/health endpoint + K8s Deployment for the consumer; per-org canary; runbook.

🤖 Generated with Claude Code

…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>

coderabbitai · 2026-06-12T09:08:21Z

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: bbf9d843-231f-459e-ad7e-9eb7505ce974

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-3541-pg-queue-consumer-launcher

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 — summary (PR #2047)

Ran six specialized agents (Code Reviewer, Silent-Failure Hunter, Type-Design Analyzer, PR Test Analyzer, Comment Analyzer, Code Simplifier). The PR is small, well-factored, and unusually well-commented; tests pass and bash -n is clean. No merge-blocking correctness bug found. Findings below are posted inline; three target pre-existing (unchanged) lines outside this diff and so are listed here only:

Pre-existing lines (cannot anchor inline):

🟠 consumer.py:248 _env() casting — cast(os.getenv(...)) raises a bare ValueError (e.g. int("abc")) with no var name / error id. Wrap the cast and name the offending WORKER_PG_QUEUE_CONSUMER_* var. (Loud is correct, but combined with the backgrounded-launch issue it gets swallowed into the log.)
🟠 consumer.py:248-249 _env typing — def _env(suffix, default, cast: type) -> object erases types at the typed PgQueueConsumer.__init__ boundary and fights the repo's strict mypy. Use a TypeVar: def _env(suffix: str, default: _T, cast: Callable[[str], _T]) -> _T.
🟡 consumer.py:215-221 poll-loop except Exception — correctly survives transient blips, but equally swallows permanent faults (schema drift, revoked grants, a _handle bug) into an infinite 0.5 Hz log-spam loop. Count consecutive failures and escalate past a threshold.

Priority order: 🔴 nohup-no-liveness (run-worker.sh:718) → 🟠 wrong-registry observability (consumer.py:185), _env casting/typing → 🟡 opt-in-hides-crash, queue-skew visibility, test gaps, call-chain comment → 🟢 pgrep || true, comment wording.

The Code Simplifier found nothing to change (the two duplication points are intentional, behavior-preserving).

- 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>

muhammad-ali-e · 2026-06-12T09:37:12Z

Review feedback addressed — `2f717c6d1`

Thanks for the thorough pass. Validity-checked each finding; here's the disposition.

Fixed

🔴 nohup no-liveness (run-worker.sh) — detached launch now verifies kill -0 after sleep 1, prints the log tail + returns non-zero on a dead fork. Applied to the shared detach path (benefits every worker).
🟠 wrong-registry observability (consumer.py) — run() logs the registered application task names at startup, so a non-empty-but-mismatched registry is diagnosable from one line (… 7 application task(s) registered: …, send_webhook_notification).
🟠 _env casting + typing — TypeVar-typed (was cast: type -> object, erasing types at the typed __init__); a bad value now raises Invalid WORKER_PG_QUEUE_CONSUMER_<X>=… instead of a context-free ValueError.
🟡 test gaps — added the guard's positive / require_tasks=False bypass / built-in-filter arms.
🟢 pgrep || true — warns on rc > 1 (operational/regex error) instead of collapsing it to "not running".
🟢 comment accuracy ×3 — build_celery_app configures but doesn't import tasks; notification is the worker that owns the leaf task send_webhook_notification; generalised the "every notification dropped" example.
SonarCloud S1192 — extracted the repeated "pg_queue_consumer" literal into a readonly constant.

Deferred (with rationale, replied inline)

🟡 opt-in-hides-crash — start-time crashes now caught by the liveness check; steady-state (died-later) detection needs the future health endpoint (already a rollout item).
🟡 queue-skew — producer/consumer queues are verified-aligned for the one migrated task; cross-queue detection belongs with the second leaf migration (don't want untested cross-queue logic now).
🟡 poll-loop escalation past a failure threshold — pre-existing 9c (UN-3539 [FEAT] PG Queue Phase 9c — consumer poll loop (claim → run → ack) #2045) behaviour, and it changes crash semantics (exit-on-sustained-fault vs log-and-continue) — worth its own discussion/PR rather than bundling here.

Re-verified live after the changes: ./run-worker.sh -r -d pg-queue-consumer → liveness passes, --status shows RUNNING, startup log lists the 7 notification tasks. Consumer test suite green.

greptile-apps · 2026-06-12T10:01:10Z

Greptile Summary

This PR wires the PG-queue consumer into the normal worker launch flow (run-worker.sh) and adds a startup guard that refuses to run against an empty task registry, preventing silent message loss from a mis-bootstrapped worker app.

__main__.py overrides WORKER_TYPE before import worker so the right source worker's tasks are loaded; the env mutation is safely gated behind if __name__ == "__main__", addressing the previous review comment about module-level side-effects.
consumer.py gains a require_tasks guard, a startup log of registered task names (to catch a wrong-type bootstrap that slips past the empty-registry check), and a properly typed _env helper.
run-worker.sh adds pg-queue-consumer as an opt-in worker type with a dedicated pgrep pattern for python -m invocations, an absolute log path so -L/-C work regardless of cwd, and an OPTIN_WORKERS map that suppresses spurious STOPPED status after an all launch.

Confidence Score: 5/5

Safe to merge — the consumer is opt-in, default all behaviour is unchanged, and the startup guard makes misconfiguration loud rather than silent.

All three changed subsystems are coherent: the __main__.py env override is correctly guarded, the require_tasks guard fires before any polling begins and is bypassed by the test suite via current_app with the module-level @shared_task tasks, and the shell wiring (pgrep pattern, absolute log path, OPTIN_WORKERS) is consistent with how pluggable workers are handled. The previous review comments about module-level side-effects and the missing usage-help entry have both been addressed in this revision.

No files require special attention.

Important Files Changed

Filename	Overview
workers/pg_queue_consumer/main.py	New entry-point that overrides WORKER_TYPE before importing `worker`, guarded behind `if __name__ == "__main__"` to prevent side-effects on import; addresses the previous review comment about module-level env mutation.
workers/pg_queue_consumer/init.py	New package marker with docstring explaining the consumer's bootstrap requirements and launch path; no logic.
workers/queue_backend/pg_queue/consumer.py	Adds `require_tasks` startup guard, startup-log of registered task names, typed `_env` helper, and improved TypeVar-based type inference; all changes are well-scoped and the guard correctly excludes `celery.*` builtins.
workers/run-worker.sh	Wires pg-queue-consumer as an opt-in worker type: correct pgrep pattern for `python -m` invocation, absolute log path fix, `OPTIN_WORKERS` map to suppress spurious STOPPED status, and best-effort crash detection on detach.
workers/tests/test_pg_queue_consumer.py	Adds four tests covering: empty-registry refusal, positive start with registered tasks via `current_app`, `require_tasks=False` bypass, and builtin-exclusion in `_registered_task_count`; correctly relies on module-level `@shared_task` tasks for the positive arm.

Sequence Diagram

sequenceDiagram
    participant Shell as run-worker.sh
    participant UV as uv run python
    participant Main as pg_queue_consumer.__main__
    participant Worker as worker.py (bootstrap)
    participant Consumer as PgQueueConsumer.run()
    participant PG as pg_queue_message

    Shell->>Shell: "export WORKER_TYPE=pg_queue_consumer"
    Shell->>Shell: "export WORKER_PG_QUEUE_CONSUMER_QUEUE=notifications"
    Shell->>Shell: "export WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE=notification"
    Shell->>UV: uv run python -m pg_queue_consumer
    UV->>Main: "__name__ == "__main__" → _bootstrap_and_run()"
    Main->>Main: "os.environ["WORKER_TYPE"] = "notification""
    Main->>Worker: import worker (registers notification tasks)
    Worker-->>Main: tasks registered in current_app
    Main->>Consumer: main() → PgQueueConsumer(...).run()
    Consumer->>Consumer: "require_tasks guard: _registered_task_count() > 0?"
    alt registry empty (wrong bootstrap)
        Consumer-->>Shell: RuntimeError (fail loud)
    else tasks present
        Consumer->>Consumer: install signal handlers
        loop poll loop
            Consumer->>PG: SKIP LOCKED claim batch
            PG-->>Consumer: messages (or empty)
            Consumer->>Consumer: task.apply() in-process
            Consumer->>PG: DELETE msg_id (ack)
        end
    end

_{Reviews (2): Last reviewed commit: "UN-3541 [FIX] Address Greptile review fe..." | Re-trigger Greptile}

- __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>

muhammad-ali-e · 2026-06-12T10:05:07Z

Greptile feedback addressed — `5bec0f8b4`

Thanks @greptile-apps (confidence 4/5). All three P2 findings handled:

Module-level side-effects in __main__.py — moved the WORKER_TYPE mutation + import worker into a guarded _bootstrap_and_run() called only under if __name__ == "__main__":. Accidental imports are now inert (verified: import leaves WORKER_TYPE untouched and doesn't bootstrap worker; python -m path unchanged).
usage() help text (outside-diff comment) — added pg-queue-consumer to the worker-type list plus a note documenting the WORKER_PG_QUEUE_CONSUMER_WORKER_TYPE / _QUEUE overrides.
Liveness check scope — kept general but clarified in the comment that it's a best-effort fast-fail for immediate (sub-second) crashes (import/config faults, which affect any worker), explicitly not a connectivity check. Left the sleep at 1s (bumping it trades latency for only partial slow-failure coverage, which needs the real health endpoint); all is unaffected since those subshells overlap the 1s with the inter-launch sleep. Rationale on the thread.

Re-verified live after the changes: python -m pg_queue_consumer boots the 7 notification tasks, --status → RUNNING, --help lists the worker.

greptile-apps · 2026-06-12T10:05:38Z

All three issues are well-addressed. The __main__.py fix is clean — _bootstrap_and_run() correctly encapsulates both the env mutation and the import side-effect, and the guard ensures tests/IDEs importing the module stay inert. The docstring and inline comment both do a good job explaining the "why" for the os.environ overwrite (vs setdefault).

No further concerns on these changes. The PR looks good from a review standpoint on these items.

sonarqubecloud · 2026-06-12T10:06:01Z

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 09:53

greptile-apps Bot reviewed Jun 12, 2026

View reviewed changes

Comment thread workers/run-worker.sh

Comment thread workers/pg_queue_consumer/__main__.py Outdated

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

muhammad-ali-e deleted the UN-3541-pg-queue-consumer-launcher branch June 12, 2026 10:15

muhammad-ali-e mentioned this pull request Jun 12, 2026

UN-3544 [FEAT] PG-queue consumer liveness endpoint (poll-loop heartbeat) #2051

Merged

Uh oh!

Conversation

muhammad-ali-e commented Jun 12, 2026

Why

What

Integration fixes found during live dev-test

Testing

Out of scope (rollout)

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 — summary (PR #2047)

Uh oh!

Uh oh!

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 — 2f717c6d1

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 — 5bec0f8b4

Uh oh!

greptile-apps Bot commented Jun 12, 2026

Uh oh!

sonarqubecloud Bot commented Jun 12, 2026

Quality Gate passed

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

coderabbitai Bot commented Jun 12, 2026 •

edited

Loading

Review feedback addressed — `2f717c6d1`

greptile-apps Bot commented Jun 12, 2026 •

edited

Loading

Greptile feedback addressed — `5bec0f8b4`