UN-3695 [FIX] PG consumer — renewable lease (dead worker redelivers in minutes, not the full VT)#2154
Conversation
…rs in minutes, not the full VT The claim was taken for the full visibility timeout (VT=9060s / 2.5h for file-processing), so a HARD-killed worker (OOM / node loss) stranded its in-flight message for ~2.5h before the vt expired and it redelivered. PR A fixed GRACEFUL kills (drain on SIGTERM); this covers the hard-kill path. - New LEASE_SECONDS (default 120): the message is claimed for the short lease, and a daemon thread renews it (client.set_vt) every ~lease/3 while the task runs. A live-but-slow task keeps its claim; a dead worker's renewal stops with the process so its lease expires in ~lease → redelivery in minutes. The lease is capped at vt_seconds; the renewal uses its OWN connection (the main thread owns the claim/ack connection) and is best-effort (a transient set_vt failure retries within the ~2x slack the lease/3 interval leaves before expiry; a False return = row already gone → stop). - VT_SECONDS is retained as the drain / max-runtime bound (it drives PR A's grace and the chart's health-stale / guards) — it is no longer the claim window. The claim now uses LEASE_SECONDS. - The renewal thread is started only around the task run, joined before the ack so the two never race the row, and creates its connection only on the first actual renewal (a task shorter than the interval opens no second connection). Paired chart change (unstract-cloud): sets WORKER_PG_QUEUE_CONSUMER_LEASE_SECONDS + a render guard LEASE <= VT. Tests: lease cap / renew-interval / claim-uses-lease / renew-until-stopped / stop-on-row-gone / swallow-transient-error / ctx-sets-stop-on-exit. 78 consumer tests pass. End-to-end dead-worker redelivery timing validated in the integration env. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
…Sonar S108) The context-manager test entered `with c._lease_renewal(42): pass` — an empty nested block (python:S108). Replace the bare pass with an observable statement (entered = True) asserted after, so the block is non-empty and still verifies the CM body runs while the renewal thread is live. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
muhammad-ali-e
left a comment
muhammad-ali-e
left a comment
There was a problem hiding this comment.
PR Review Toolkit — automated review (6 agents)
The renewable-lease change is well-structured overall: the daemon-thread + Event + context-manager shape is idiomatic, the lazy second connection and join-before-ack are well-reasoned, and the happy-path tests are solid. Findings are grouped below and posted as inline comments.
High
- BATCH>1 tail double-processing (
consumer.py:302) + now-stale mitigation comment (consumer.py:59) - Renewal error handling: over-broad
except+ no escalation on prolonged failure (consumer.py:358) set_vt→Falsemid-run is an unlogged double-run signal (consumer.py:356)
Medium
- Wedged-join not logged + shared
_renew_clientreuse race (consumer.py:339) - Silent
lease>vtcap + factually wrong 'fires rarely' comment (consumer.py:265) _renew_clientnever closed (consumer.py:270)- Test gaps: lazy-init,
_handleintegration, close, boundaries (test file)
Low
- 'never race' docstring overstated (
consumer.py:325) - Missing
-> Iterator[None](consumer.py:316) - README: LEASE default vs default VT (
README.md:31); lease is not the max-runtime bound (README.md:78)
The default config (BATCH=1) is safe from the top item; the batch and error-handling findings bite specifically under the file-processing profile this feature targets (large VT, batching, long tasks).
| try: | ||
| messages = self._client.read( | ||
| queue_name, vt_seconds=self.vt_seconds, qty=self.batch_size | ||
| queue_name, vt_seconds=self.lease_seconds, qty=self.batch_size |
There was a problem hiding this comment.
[High] BATCH>1 tail messages can double-process. read() now claims the whole batch for lease_seconds, but _handle processes messages sequentially and _lease_renewal only renews the in-flight msg_id. Messages #2..N sit in the local list unrenewed; if the head task runs longer than lease_seconds (routine for file-processing) their vt expires, another consumer reclaims them, and they run twice. Before this PR the tail was hidden for the full vt_seconds (~2.5h). Safe at the default BATCH=1, but BATCH is a supported knob on exactly the long-task/large-VT profile. Fix options: renew every still-unacked msg_id in the batch; or claim the tail for vt_seconds; or force batch_size=1 when lease_seconds < vt_seconds. Coupled to the stale comment at line 59.
| @@ -57,6 +59,15 @@ | |||
| # raise it, keep vt_seconds > batch_size x worst-case task duration. | |||
There was a problem hiding this comment.
[High] This mitigation note is now wrong and will mislead operators. The batch claim window is lease_seconds (line 302), not vt_seconds, so 'keep vt_seconds > batch_size × worst-case task duration' no longer protects the batch tail — raising VT_SECONDS does nothing for queued-but-unstarted messages once the lease caps the claim window. Update to reference LEASE_SECONDS and tie it to the batch-renewal fix (line 302).
| self._renew_client = PgQueueClient() | ||
| if not self._renew_client.set_vt(msg_id, self.lease_seconds): | ||
| return # row gone — nothing left to renew | ||
| except Exception: |
There was a problem hiding this comment.
[High] except Exception is too broad, and repeated failures never escalate. (1) A deterministic bug (an AttributeError from a rename, or set_vt's own ValueError) is caught every tick and logged as a self-healing 'blip' — it never self-heals, the lease is never renewed, and every long task silently double-runs. Narrow the catch to the connection-dead types (_CONN_DEAD_ERRORS) and log unexpected types at ERROR, then stop retrying. (2) Even for genuine transient failures: if the DB is down longer than lease_seconds the lease is lost and the task double-runs, yet every tick emits the same reassuring WARNING ('a dead connection self-heals on the next tick'). Track last-success time and escalate to ERROR with a distinct, greppable message once the failure window exceeds lease_seconds (the point past which a double-run is provably possible).
| # task shorter than the interval opens no second connection. | ||
| if self._renew_client is None: | ||
| self._renew_client = PgQueueClient() | ||
| if not self._renew_client.set_vt(msg_id, self.lease_seconds): |
There was a problem hiding this comment.
[High] set_vt returning False mid-run is an unlogged double-run signal. The renewal thread is joined before the ack (line 339; the delete happens after the with block exits), so it can never observe our own ack — the docstring's 'acked by us' case is impossible in this thread. Mid-run, False means the lease already expired and another worker reclaimed the row, i.e. the task is double-running right now. Yet the loop just returns silently with no log. Emit a WARNING (arguably ERROR) before returning, and correct the docstring at lines 347-348 — 'acked by us' should read 'expired + reclaimed → likely double-running'.
| yield | ||
| finally: | ||
| stop.set() | ||
| thread.join(timeout=_LEASE_JOIN_TIMEOUT_SECONDS) |
There was a problem hiding this comment.
[Medium] Wedged renewal thread: the join can time out silently, and the shared _renew_client can then be used by two threads. If set_vt is blocked in a hung DB call, join(timeout=10) returns with the thread still alive and execution falls through to the ack with no is_alive() check and no log. Compounding hazard: the next message starts a new renewal thread that reuses the same self._renew_client connection (line 355) — two threads on one non-thread-safe psycopg connection → protocol corruption that surfaces later on an unrelated statement. Add an if thread.is_alive(): ERROR log after the join, and consider not sharing _renew_client across tasks while a prior thread may still be alive.
| dies with it → the lease expires in ~``lease_seconds`` → the message | ||
| redelivers in minutes instead of the full VT. The renewal uses its OWN | ||
| connection (``_renew_client``); the main thread owns ``_client`` for the ack, | ||
| and the thread is joined here before that ack so the two never race the row. |
There was a problem hiding this comment.
[Low] Docstring overstates the guarantee. 'the thread is joined here before that ack so the two never race the row' — the join has a 10s timeout (line 339); on timeout the thread is still alive and set_vt can run after the delete. The race is benign (set_vt on a deleted row updates 0 rows → False), but 'never race' contradicts the honest constant comment at lines 66-67. Soften to: 'joined best-effort, bounded by _LEASE_JOIN_TIMEOUT_SECONDS; a late set_vt on the acked row no-ops.'
| return total | ||
|
|
||
| @contextlib.contextmanager | ||
| def _lease_renewal(self, msg_id: int): |
There was a problem hiding this comment.
[Low] Missing return-type annotation. This is the only method in the file without one. A @contextmanager generator should be -> Iterator[None] (add Iterator to the collections.abc import at line 30).
| | `CONCURRENCY` | Prefork consumer children per pod (1 = plain single process) | `1` | | ||
| | `VT_SECONDS` | **Visibility timeout** — how long a claimed message stays hidden before it can redeliver | `30` (chart: `9060` for file-processing ≈ 2.5h) | | ||
| | `VT_SECONDS` | Since PR C, the **drain / max-runtime bound** (drives `SHUTDOWN_GRACE`, `HEALTH_STALE`, the chart guards) — the *claim* window is `LEASE_SECONDS`, not this | `30` (chart: `9060` for file-processing ≈ 2.5h) | | ||
| | `LEASE_SECONDS` | **Renewable claim window** — a claimed message is hidden only this long and renewed every ~LEASE/3 while the task runs; a dead worker's claim expires in ~LEASE → fast redelivery (capped at VT) | `120` | |
There was a problem hiding this comment.
[Low] LEASE_SECONDS default 120 is misleading under the default VT. Because the effective lease is min(LEASE, VT), with the default VT=30 the lease clamps to 30 and the renew interval to 10s — not the 120 / ~40s a reader infers from this row. The minutes-not-hours benefit only materializes once VT is raised (the 9060 file-processing chart). Add a parenthetical, e.g. 120 (effective = min(LEASE, VT); with the default VT=30 this clamps to 30).
| **renews** it (`set_vt`) every ~`LEASE/3` while the task runs. A live-but-slow task | ||
| keeps its claim; a **dead** worker's renewal stops, so its `vt` expires in ~`LEASE` → | ||
| redelivery in **minutes, not hours**. `VT_SECONDS` is retained as the *drain / | ||
| max-runtime bound* (grace, health-stale), and the lease is capped at it. The renewal |
There was a problem hiding this comment.
[Low] The lease does not self-cap a live-but-hung task. The renewal loop has no absolute deadline — a live-but-wedged task renews forever and is never redelivered, so VT_SECONDS is not a max-runtime bound via the lease. In practice task.apply runs synchronously on the poll thread, so a hung task stalls the consumer and the external health-stale / liveness probe kills the pod (process death stops renewal → redelivery). Consider rewording so operators understand the bound is enforced by the probe killing the pod, not by the lease self-capping.
| ) | ||
|
|
||
|
|
||
| class TestLeaseRenewal: |
There was a problem hiding this comment.
[Medium] Test gaps in the new suite. Every test pre-sets _renew_client = MagicMock(), so the load-bearing paths are untested: (1) lazy PgQueueClient() creation — that it's built once and not built for a sub-interval task (the 'no second connection' guarantee); (2) _handle integration — that task.apply runs inside _lease_renewal and the thread is joined before the ack (deleting the with wrapper wouldn't fail any current test); (3) the join-timeout / wedged-thread path; (4) _renew_client close on shutdown (see consumer.py:270); (5) boundary values — lease_seconds<=0 rejected at construction, lease == vt, and the tiny-lease max(1, //3) no-slack edge. Also stop.wait is a bare mock, so the timing invariant _lease_renew_interval < lease_seconds is never asserted.
muhammad-ali-e
left a comment
muhammad-ali-e
left a comment
There was a problem hiding this comment.
PR Review Toolkit — automated multi-agent review
Ran 6 specialized agents (code review, silent-failure hunt, type design, test coverage, comment accuracy, simplification) over the renewable-lease change. The design is sound and the happy path (live worker renews / dead worker's lease expires) is correct and well-tested. Findings below are grouped as inline comments.
Priority order:
- Important — shared
_renew_clientconcurrency race when the bounded join times out (2 agents found this independently). - Important — three silent/under-observed unhappy-path renewal failures (row-gone return, sustained-failure escalation, wedged-thread) — a lost lease here means duplicate execution / double-billing, so these deserve a log + metric.
- Important — an inaccurate code comment + misleading README/default interaction around
min(lease, vt). - Suggestions/nits — metrics wiring, connection close, comment precision, return annotation.
- Tests — integration/ordering/edge-case coverage gaps.
Note: one agent's finding that a dead _renew_client "never rebuilds / self-heal is false" was verified and dropped — _cursor() (client.py:219-222) discards a dead owned connection, so set_vt does reconnect on the next tick; the comment is accurate.
| yield | ||
| finally: | ||
| stop.set() | ||
| thread.join(timeout=_LEASE_JOIN_TIMEOUT_SECONDS) |
There was a problem hiding this comment.
[Important] Bounded join can leave a wedged renewal thread sharing _renew_client with the next task's thread → concurrent use of one psycopg2 connection.
If a renewal thread is blocked inside set_vt on a stalled connection (DB failover / PgBouncer reap / network partition — exactly the window this feature targets) for >10s, join(timeout=…) returns silently and _handle proceeds. The main loop then advances to the next message, whose renewal thread reuses the same self._renew_client (if self._renew_client is None at :354 is already non-None). Two threads issuing set_vt on one libpq connection is undefined behavior (protocol desync), which can then make renewals fail and reintroduce the very redelivery this PR prevents. The docstring's "joined here before that ack so the two never race the row" (:325) also does not hold once the join times out.
Suggested fix: give each renewal its own PgQueueClient scoped to _lease_renewal (and close() it), or guard _renew_client use with a threading.Lock. Also log + increment a metric when thread.is_alive() after the join — an abandoned thread is currently completely silent.
| if self._renew_client is None: | ||
| self._renew_client = PgQueueClient() | ||
| if not self._renew_client.set_vt(msg_id, self.lease_seconds): | ||
| return # row gone — nothing left to renew |
There was a problem hiding this comment.
[Important] set_vt → False (row gone) returns with zero logging — the most definitive double-run signal is silent.
During an in-flight renewal the ack has not run yet (the thread is joined before the ack at :504). Verified against client.py: read() reclaim only UPDATEs vt/read_ct, it does not delete — so set_vt returns False only after the row was actually DELETED, i.e. another worker already reclaimed and acked it → this task is being/has-been double-processed. Contrast the ack path at :504-510, which logs a WARNING for the same condition. Here it's a bare return.
Suggested fix: logger.error(...) naming msg_id + a lease_lost counter before returning.
Also: the docstring at :347 lists "acked by us" as a cause of False — that's impossible inside the loop (ack happens after join), and "expired + reclaimed" does not by itself return False (reclaim doesn't delete). Reword to: "False = the row was deleted (acked/dropped by its current owner)."
| if not self._renew_client.set_vt(msg_id, self.lease_seconds): | ||
| return # row gone — nothing left to renew | ||
| except Exception: | ||
| logger.warning( |
There was a problem hiding this comment.
[Important] The loop can't distinguish a transient blip from a definitively-lost lease — sustained failure silently allows a concurrent double-run.
Every set_vt exception logs an identical WARNING and loops with no state about how long renewals have been failing. If set_vt fails on every tick for longer than lease_seconds (dead conn / failover), the lease expires while task.apply is still running, another worker reclaims the row, and the task runs twice — the worker that lost the lease emits only look-alike WARNINGs, never escalates. Given this subsystem's own delete() docstring ties a lost lease to duplicate webhooks + subscription-usage billing, a blip and a total loss should not be indistinguishable in the logs.
Suggested fix: track last_ok = time.monotonic(); once now - last_ok >= lease_seconds, escalate to logger.error + a lease_renewal_giveup metric so it pages, while keeping the sub-lease case at WARNING.
| # Renewable lease (UN-3695 PR C): claim for lease_seconds (short) and renew | ||
| # every ~lease/3 while a task runs. Capped at vt_seconds so the lease is | ||
| # genuinely the shorter window; setting lease >= vt degrades to the old | ||
| # single-long-claim behaviour (renewal fires rarely). |
There was a problem hiding this comment.
[Important] This comment is factually wrong. After self.lease_seconds = min(lease_seconds, vt_seconds) (:265) the effective lease can never exceed vt, so _lease_renew_interval = vt//3 and renewal fires every vt/3 — frequently, re-parking on every tick. That is neither "single-long-claim behaviour" (which never renews) nor "renewal fires rarely."
Suggested: "setting lease >= vt just clamps the lease to vt; renewal still fires every vt/3 (redundantly, each extending by vt)."
| # every ~lease/3 while a task runs. Capped at vt_seconds so the lease is | ||
| # genuinely the shorter window; setting lease >= vt degrades to the old | ||
| # single-long-claim behaviour (renewal fires rarely). | ||
| self.lease_seconds = min(lease_seconds, vt_seconds) |
There was a problem hiding this comment.
[Minor] With the shipped code defaults the feature is inert. _DEFAULT_LEASE_SECONDS = 120 (:67) but _DEFAULT_VT_SECONDS = 30 (:66-ish), so a default-constructed consumer gets min(120, 30) = 30 — the advertised 120s lease is silently discarded and lease == vt. It only does something once VT_SECONDS is raised well above the lease (which the chart does in prod: VT=9060). This is a dev/local sharp edge, not a prod bug — but worth either making the defaults coherent (lower the default lease below the default VT, or raise the default VT to match its new "max-runtime bound" role) or logger.warning when lease_seconds > vt_seconds so the clamp is visible. Note the constructor rejects backoff_max < poll_interval explicitly (:236) yet clamps this cross-field invariant silently — inconsistent.
| self._lease_renew_interval = max(1, self.lease_seconds // 3) | ||
| # Second client (own connection) for lease renewals — the main thread owns | ||
| # self._client for the read/ack, so a renewal must not share that connection. | ||
| # Lazily created on the first task run. |
There was a problem hiding this comment.
[Nit] Imprecise + duplicated comment. "Lazily created on the first task run" is slightly wrong — it's created on the first renewal (:352-353 states this accurately), which only happens when a task outlives the interval; a short task opens no second connection. Suggest: "Lazily created on the first lease renewal (never for tasks shorter than the interval)." Keep only one authoritative statement of this — :352-353 already has it.
| return total | ||
|
|
||
| @contextlib.contextmanager | ||
| def _lease_renewal(self, msg_id: int): |
There was a problem hiding this comment.
[Nit] Missing return-type annotation. The rest of the file annotates returns (poll_once -> int, _renew_lease_loop -> None); this contextmanager omits it. Add -> Iterator[None] (from collections.abc import Iterator) so the generator contract is expressed and type-checkable.
| ) | ||
|
|
||
|
|
||
| class TestLeaseRenewal: |
There was a problem hiding this comment.
[Tests] Good loop-logic coverage; the gaps are at the seams. The state-machine tests (cap, interval, claim-uses-lease, renew-until-stopped, stop-on-row-gone, swallow-and-retry, ctx stop-on-exit) are solid. Missing, roughly in priority order:
- Integration — nothing drives
_handle/poll_onceto asserttask.applyis actually wrapped in_lease_renewal(msg_id)(consumer.py:424-430). A refactor dropping thewithwould keep all 7 tests green. - Ordering — the "joined before ack" safety claim (thread stopped/joined at :338-339 before
deleteat :504) is unasserted. - Lazy
_renew_client— every loop test pre-setsc._renew_client = MagicMock(), so theif self._renew_client is None: PgQueueClient()branch (:354-355) and the "no connection for short tasks" promise have zero coverage. lease_seconds <= 0validation —TestConstruction.test_rejects_non_positive_paramsenumerates every other positive-int param but omitslease_seconds; add0and-5cases (trivial, guards the zero-lease continuous-double-run foot-gun).LEASE_SECONDSenv wiring —build_consumer_from_env(consumer.py:999) has no test at all; the operator-facing knob could be mis-spelled/dropped with CI green.- Default cap 120→30 and the
max(1, lease//3)floor for tiny leases (1/2 → 1) and thelease == vtboundary are untested. - Sustained
set_vtfailure past the lease (the double-run boundary) is untested.
| |---|---|---| | ||
| | `CONCURRENCY` | Prefork consumer children per pod (1 = plain single process) | `1` | | ||
| | `VT_SECONDS` | **Visibility timeout** — how long a claimed message stays hidden before it can redeliver | `30` (chart: `9060` for file-processing ≈ 2.5h) | | ||
| | `VT_SECONDS` | Since PR C, the **drain / max-runtime bound** (drives `SHUTDOWN_GRACE`, `HEALTH_STALE`, the chart guards) — the *claim* window is `LEASE_SECONDS`, not this | `30` (chart: `9060` for file-processing ≈ 2.5h) | |
There was a problem hiding this comment.
[Docs] These two rows are misleading for the default/local config. VT_SECONDS row asserts "the claim window is LEASE_SECONDS, not this" — but with the default VT=30 < LEASE=120 the effective claim window is VT (30), because of min(lease, vt). And the LEASE_SECONDS Default column says 120 while the effective default is 30. Suggest: VT row → "the claim window is min(LEASE_SECONDS, VT_SECONDS)"; LEASE row Default → "120 (effective = min(LEASE, VT); with the default VT=30, 30)".
Also: "Since PR C" / "UN-3695 PR C" shorthand (:30, :73, and consumer.py:62/261/317/423) will rot after merge — no artifact called "PR C" survives. Prefer the UN-3695 Jira ref or describe the behavior directly, especially in the README where a standalone reader can't resolve it.
| uses its own DB connection and is best-effort (a transient failure retries within the | ||
| `~2×` slack the `LEASE/3` interval leaves before expiry). | ||
|
|
||
| **Claim** — an atomic `SELECT … FOR UPDATE SKIP LOCKED` that hides up to `BATCH` |
There was a problem hiding this comment.
[Docs] Stale glossary entry. The "Claim" definition still says it hides rows "for VT seconds," but poll_once now reads with vt_seconds=self.lease_seconds (consumer.py:302). Update to "...hides up to BATCH ready rows for the lease window (min(LEASE, VT))..." so it matches the PR-C model introduced just above.
…servable failures, batch guard Review round on the renewable-lease change (#2154): - Race: the renewal thread no longer shares self._renew_client. It owns a thread-local connection built lazily on the first renewal and closed in a finally, so a join-timeout wedge can't leave the next message's thread reusing a half-used psycopg2 connection. _make_renew_client() is the (patchable) factory. - Wedge is no longer silent: after the bounded join, a still-alive thread is logged at ERROR (connection abandoned until process exit). - set_vt -> False mid-run is a double-run signal (the ack runs only after this thread is joined, so the row can only be gone if another consumer reclaimed+acked it) — now logged, and the docstring corrected (was wrongly "acked by us"). - Narrowed the renewal except to CONN_DEAD_ERRORS (a non-connection error propagates instead of being swallowed as a "self-healing blip" forever) and added escalation: once renewal keeps failing past lease_seconds the lease is genuinely lost → ERROR. - BATCH_SIZE forced to 1 when lease < vt: renewal only covers the in-flight message, so a batch tail could lapse and double-run. All shipped workers are batch=1; this guards the knob. - Clamp is loud: lease > vt now warns (was a silent min()); comment corrected. - Iterator return annotation on the context manager; "PR C" shorthand → UN-3695. - README: effective claim window is min(LEASE, VT); lease is not a hard runtime cap (liveness probe is the backstop); batch-forced-to-1 note. - Tests: +9 (loud clamp / tiny-interval floor / non-positive reject / batch-forced / batch-kept-at-lease==vt / env wiring / renew+close / no-conn-for-sub-interval / row-gone-logs / conn-death-retry-then-escalate / non-conn-propagates / wedged-log / _handle-wraps-in-lease). 87 consumer tests pass. Deferred: lease metrics counters (ConsumerMetrics isn't wired to the consumer instance) — a separate small observability change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Thanks for the thorough pass — addressed the review in Correctness / safety (the top-3 + high items)
Docs / nits
Tests (#13): +9 covering loud clamp / tiny-interval floor / non-positive reject / batch-forced / batch-kept-at- Deferred (1): lease metrics counters (#8) — The one round-1 finding your verifier already dropped ("dead client never self-heals") stays dropped — |
…hrow pytest.raises, empty block - consumer.py: escalation log uses logger.exception() (ERROR + traceback) instead of logger.error(..., exc_info=True) in the renewal except block. - tests: one throwing call per pytest.raises block; fill the empty _lease_renewal context-manager with-block (S108). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
| Filename | Overview |
|---|---|
| workers/queue_backend/pg_queue/consumer.py | Core change: adds lease_seconds param, _lease_renewal context manager, and _renew_lease_loop daemon thread; batch forced to 1 when lease < VT; claim uses lease_seconds instead of vt_seconds. Logic is sound: join-before-ack ensures no set_vt/delete race; lazy connection avoids wasted resources for short tasks; clamping and batch guard are correct. |
| workers/tests/test_pg_queue_consumer.py | 17 new TestLeaseRenewal tests cover: construction/clamping, env wiring, claim uses lease not VT, loop renews-until-stopped/conn-death-retry/escalation/early-stop-on-row-gone, context manager stop+join, wedged-thread log, and _handle integration guard. Deterministic control of stop.wait and time.monotonic is well-structured. |
| workers/queue_backend/pg_queue/README.md | Documentation updated to reflect LEASE_SECONDS env var, the new role of VT_SECONDS as a drain bound, and the renewable-lease glossary entry explaining the fast-redelivery mechanic. Changes are accurate and thorough. |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant M as Main Thread
participant R as Renewal Thread (daemon)
participant DB as Postgres
M->>DB: "read(vt=lease_seconds) — claim msg"
DB-->>M: QueueMessage(msg_id)
M->>R: start _lease_renewal(msg_id)
Note over R: stop.wait(lease/3) — sleeping
M->>M: task.apply() — running task
loop Every lease/3 seconds while task runs
R->>DB: set_vt(msg_id, lease_seconds)
DB-->>R: True (row still exists)
end
M-->>M: task returns
M->>R: stop.set()
R-->>R: stop.wait() returns True → exit loop
R->>DB: client.close()
M->>M: thread.join(10s) — wait for renewal to finish
M->>DB: delete(msg_id) — ack
Note over M,DB: If worker dies hard (OOM/node loss):<br/>Renewal thread dies with process.<br/>VT expires in ~lease_seconds → redelivery
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant M as Main Thread
participant R as Renewal Thread (daemon)
participant DB as Postgres
M->>DB: "read(vt=lease_seconds) — claim msg"
DB-->>M: QueueMessage(msg_id)
M->>R: start _lease_renewal(msg_id)
Note over R: stop.wait(lease/3) — sleeping
M->>M: task.apply() — running task
loop Every lease/3 seconds while task runs
R->>DB: set_vt(msg_id, lease_seconds)
DB-->>R: True (row still exists)
end
M-->>M: task returns
M->>R: stop.set()
R-->>R: stop.wait() returns True → exit loop
R->>DB: client.close()
M->>M: thread.join(10s) — wait for renewal to finish
M->>DB: delete(msg_id) — ack
Note over M,DB: If worker dies hard (OOM/node loss):<br/>Renewal thread dies with process.<br/>VT expires in ~lease_seconds → redelivery
Reviews (2): Last reviewed commit: "UN-3695 [FIX] startup log shows lease (t..." | Re-trigger Greptile
… (Greptile P2) vt_seconds is now the drain/max-runtime bound, not the claim window — the startup log showing only vt=%ss misleads an operator into thinking the claim window is 9060s when it's actually lease_seconds (e.g. 120s). Add lease=%ss. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
|
Thanks Greptile (4/5, safe-to-merge) — addressed the one P2 in |
muhammad-ali-e
merged commit f9b8da1
into
feat/UN-3445-pg-queue-integration
1 participant
What
Follow-on to the merged grace=VT fix (#2153). That covered graceful kills (drain on SIGTERM). This covers hard kills (OOM / node loss), where the worker gets no SIGTERM: the message was claimed for the full visibility timeout (VT=9060s / 2.5h for file-processing), so a hard-killed worker stranded its in-flight message for ~2.5h before the vt expired and it redelivered.
LEASE_SECONDS(default 120): the message is claimed for the short lease, and a daemon thread renews it (client.set_vt) every ~lease/3while the task runs. A live-but-slow task keeps its claim; a dead worker's renewal stops with the process, so its lease expires in ~lease → redelivery in minutes, not 2.5h.VT_SECONDSis retained as the drain / max-runtime bound — it still drives PR A's shutdown grace and the chart'shealth-stale/ render guards. It is no longer the claim window; the claim now usesLEASE_SECONDS(capped at VT).task.apply, joined before the ack so the two never race the row; it uses its own DB connection (the main thread owns the claim/ack connection); it's best-effort — a transientset_vtfailure retries within the ~2× slack thelease/3interval leaves before expiry, and aFalsereturn (row already acked/expired) stops it. The renew connection is opened only on the first actual renewal, so a task shorter than the interval opens no second connection.Why
PR A made deploy/scale-down orphans impossible, but OOM/node-death (no SIGTERM) still stranded a batch for the full VT. Shortening recovery to minutes closes the hard-kill gap and brings PG closer to Celery's broker-requeue latency.
Scope / safety
LEASE_SECONDS=120works with no chart change; the paired unstract-cloud change sets it explicitly per pg-worker + adds aLEASE ≤ VTrender guard.How tested
stop.wait): lease cap-at-VT / renew-interval (lease//3) / claim uses lease not VT / renew-until-stopped / stop-on-row-gone / swallow-transient-error-and-retry / ctx-sets-stop-on-exit. 78 consumer tests pass.Part of UN-3695. Depends on nothing; complements the merged #2153.
🤖 Generated with Claude Code