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

…Cloud)

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>

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

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