UN-3695 [FIX] PG consumer — supervisor shutdown grace = VT, single shared join deadline#2153
Conversation
…ared join deadline The prefork supervisor SIGKILLed its own children after a hardcoded 30s on shutdown (`_SHUTDOWN_GRACE_SECONDS = 30.0`). But the chart grants the pod `terminationGracePeriodSeconds = VT + buffer` (9120s for worker-pg-file-processing, VT=9060) — so on a graceful SIGTERM (deploy / HPA pod scale-down) the supervisor was knifing an in-flight batch ~300x sooner than the pod grace it lives inside, orphaning the batch (executor finished, but nothing consumed the reply / marked COMPLETE / decremented the barrier / acked the message). - `shutdown_grace_from_env()`: the per-child drain grace now defaults to the consumer VT (`WORKER_PG_QUEUE_CONSUMER_VT_SECONDS`), floored at 30 so it can never be SHORTER than before (non-regressive; OSS/dev with the default VT stays 30, fileproc gets 9060 with no chart change). Explicit `WORKER_PG_QUEUE_CONSUMER_SHUTDOWN_GRACE_SECONDS` overrides. - `_join_children` now uses a SINGLE shared deadline instead of a per-child recomputed one. Children are SIGTERM'd together and drain in parallel, so one shared window still gives each its full grace while bounding total drain to ~grace. A per-child deadline would sum to N x grace and, at grace=VT, blow past the pod grace so k8s SIGKILLs the whole pod, hard-killing siblings that were draining cleanly. PG-worker-component only — the supervisor never runs for Celery workers, so no flag branch is needed and the Celery path is untouched. Tests: shutdown_grace_from_env (VT-tracking / floor / override) + a shared-deadline regression test (38 supervisor tests). Verified e2e with real forked children: grace tracks VT and drains a live child (no premature SIGKILL); 2 wedged children SIGKILL together at the shared deadline, not 2x. 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 |
2562182 to
e61708f
Compare
Add a co-located README as a lookup dictionary for the PG-queue transport: config env vars (consumer / reaper / orchestration / routing) with defaults, a concepts glossary (visibility timeout, claim, redelivery, reaper, barrier, poison/re-park, prefork supervisor, shutdown grace, request-reply/callback, last_progress_at, leader election, fairness), and the table catalogue. Complements the 9e-design.md design doc. Docs only — no code change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
e61708f to
baa0199
Compare
muhammad-ali-e
left a comment
There was a problem hiding this comment.
Automated PR review (PR Review Toolkit: code-reviewer, silent-failure-hunter, type-design, comment, test, simplifier). The core fix is correct — grace defaults to VT floored at 30s, override honoured as-is, and the single shared deadline correctly bounds total drain to ~grace instead of N×grace. Findings below are ranked; the two [med] items are the ones worth addressing before merge. Everything else is low/nit.
| """ | ||
| from queue_backend.pg_queue.consumer import _DEFAULT_VT_SECONDS, consumer_env | ||
|
|
||
| override = consumer_env("SHUTDOWN_GRACE_SECONDS", None, float) |
There was a problem hiding this comment.
[med] SHUTDOWN_GRACE_SECONDS override is unvalidated — a bad value silently produces the exact failure UN-3695 fixes.
consumer_env(..., float) accepts negative and non-finite values, and max(0.0, override) then collapses them:
SHUTDOWN_GRACE_SECONDS=-30→max(0.0, -30)=0.0→ deadlinenow+0→ every child SIGKILLed with zero drain on the next scale-down (the orphaned-batch bug this PR exists to prevent), with no log.nan→max(0.0, nan)=0.0→ same immediate kill.inf/1e999→max(0.0, inf)=inf→_join_childrendeadline isnow+inf→ shutdown hangs forever until k8s hard-kills the whole pod.
A negative/non-finite drain budget is never intentional. Sibling concurrency_from_env() already raises on n < 1 — mirror that here:
override: float | None = consumer_env("SHUTDOWN_GRACE_SECONDS", None, float)
if override is not None:
if not math.isfinite(override) or override < 0:
raise ValueError(
f"WORKER_PG_QUEUE_CONSUMER_SHUTDOWN_GRACE_SECONDS must be a finite "
f"number >= 0, got {override!r}"
)
return override(The override: annotation also restores the X | None intent-declaration convention this module documents at consumer.py:858.)
| deadline = time.monotonic() + grace_seconds | ||
| for slot, pid in fleet.alive_items(): | ||
| if _wait_for_exit(pid, time.monotonic() + grace_seconds): | ||
| if _wait_for_exit(pid, deadline): |
There was a problem hiding this comment.
[med] New regression: a clean-exited child iterated after the shared deadline elapses is spuriously SIGKILLed and logged as "did not drain".
With one wedged child consuming the whole shared window, every subsequent child hits _wait_for_exit(pid, deadline) with deadline already in the past. _wait_for_exit's loop is while time.monotonic() < deadline: — so it returns False without ever calling waitpid, even for a child that exited cleanly at T+5s. _join_children then SIGKILLs it and logs "did not drain within the shared %ss grace".
The SIGKILL to the (already-zombie) child is harmless, but the WARNING is a false "batch was hard-killed" alarm — ironic for the PR whose whole point is to stop hard-killing in-flight batches, and it can trip exactly that alert. This is new: the old per-child code recomputed time.monotonic() + grace per iteration, so the deadline was never already-elapsed at entry.
Fix in _wait_for_exit — poll waitpid at least once regardless of the deadline:
def _wait_for_exit(pid: int, deadline: float) -> bool:
while True:
try:
reaped, _status = os.waitpid(pid, os.WNOHANG)
except ChildProcessError:
return True
if reaped != 0:
return True
if time.monotonic() >= deadline:
return False
time.sleep(0.1)| fleet = _Fleet(concurrency) | ||
| grace_seconds = shutdown_grace_from_env() | ||
| logger.info( | ||
| "PG-queue consumer supervisor: shutdown drain grace = %.0fs/child", grace_seconds |
There was a problem hiding this comment.
[low] Startup log says %.0fs/child but the drain is now a single shared total, not per-child.
After this PR grace_seconds bounds the total drain to ~grace across all children (see the _join_children docstring, which explicitly rejects the per-child model). An operator reading grace = 9060s/child with concurrency=4 will mis-model the total window as 4×9060s — the precise per-child-vs-shared misconception the rework fixes. Drop the /child:
"PG-queue consumer supervisor: shutdown drain grace = %.0fs (shared across children)", grace_seconds|
|
||
|
|
||
| def shutdown_grace_from_env() -> float: | ||
| """Per-child graceful-drain budget (seconds) on shutdown, before SIGKILL. |
There was a problem hiding this comment.
[low] "Per-child" wording left stale after the switch to a shared deadline (3 sites).
This docstring's first line still calls the value a "Per-child graceful-drain budget", and so does the constant comment at line 66 and the README table row at README.md:25 — yet _join_children now applies it as a single shared total and its own docstring explicitly warns that a per-child deadline would be a bug. Same knob described two contradictory ways within one PR. Suggest a find-and-replace to "graceful-drain budget, shared across all children on shutdown" at all three sites (the _join_children docstring and the SIGKILL warning at line 445 are already correct — align the rest to them).
| if override is not None: | ||
| return max(0.0, override) | ||
| vt = consumer_env("VT_SECONDS", _DEFAULT_VT_SECONDS, int) | ||
| return max(float(_DEFAULT_SHUTDOWN_GRACE_SECONDS), float(vt)) |
There was a problem hiding this comment.
[nit] Redundant float() on an already-float constant. _DEFAULT_SHUTDOWN_GRACE_SECONDS is 30.0 (line 71), so float(...) around it is a no-op. float(vt) is the meaningful cast (vt is an int):
return max(_DEFAULT_SHUTDOWN_GRACE_SECONDS, float(vt))| |---|---|---| | ||
| | `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) | | ||
| | `SHUTDOWN_GRACE_SECONDS` | Per-child graceful-drain budget on SIGTERM before SIGKILL | `= VT`, floored at `30` | |
There was a problem hiding this comment.
[low] This row contradicts the glossary in the same file. It calls SHUTDOWN_GRACE_SECONDS a "Per-child graceful-drain budget", but the "Shutdown grace" glossary entry at line 87 correctly says the supervisor drains "using a single shared deadline." Reword to Graceful-drain budget (shared across all children) on SIGTERM before SIGKILL so the two agree.
| kill.assert_called_once_with(111, _signal.SIGKILL) | ||
| waitpid.assert_called_once_with(111, 0) | ||
|
|
||
| def test_shared_deadline_not_per_child(self): |
There was a problem hiding this comment.
[test] Coverage gaps around the change's load-bearing paths.
- Multi-child straggler on the shared deadline is untested.
test_shared_deadline_not_per_childonly exercises the clean-exit path (_capturealways returnsTrue);test_straggler_is_sigkilled_and_reapeduses a single child. A regression re-introducing a per-child deadline inside the SIGKILL loop would pass both. Add a 3-child, all-False(none drain) case asserting one shared deadline andkill/waitpidcalled 3×. - This assertion is probabilistic.
len(set(deadlines)) == 1only distinguishes shared-vs-per-child iftime.monotonic()ticks between iterations; with_wait_for_exitpatched to return instantly a per-child bug could yield 3 identical values on a coarse clock and pass. Harden by patchingtime.monotonicwith an advancingside_effect=[100.0, 200.0, 300.0, 400.0]and assertingdeadlines == [105.0, 105.0, 105.0]. - Negative override branch untested — add
SHUTDOWN_GRACE_SECONDS=-5(pairs with the validation fix suggested onsupervisor.py:118). - Malformed env is untested — assert
shutdown_grace_from_env()raisesValueErrormatching the offending var name forSHUTDOWN_GRACE_SECONDS=abcandVT_SECONDS=9060s, pinning the fail-fast-at-startup contract.
|
Thanks — all 7 addressed in
SonarCloud's "C Reliability on New Code" should clear on this commit — it was the same override/hang class + the lambda idiom, both now fixed. |
…adline false-alarm, docs
Review round on the grace=VT / shared-deadline change:
- shutdown_grace_from_env: reject a negative / non-finite SHUTDOWN_GRACE_SECONDS
override (was silently collapsed by max(0.0, ...) -> 0 = immediate SIGKILL, or inf
= shutdown hang) — fail fast like concurrency_from_env; annotate override: float|None;
drop the redundant float() on the already-float default.
- _wait_for_exit: poll waitpid at least once regardless of the deadline. On the
single SHARED deadline a child iterated after the window elapsed was reported
'did not drain' and SIGKILLed — a false hard-kill alarm — even though it exited.
- Wording: the grace is a single shared total, not per-child — corrected the
docstring, constant comment, startup log ('shared across children'), and the
README config row to match the _join_children docstring.
- Tests: shared-deadline hardened with an advancing time.monotonic (exact
105/105/105); a 3-child all-wedged straggler case (kill+waitpid x3, one shared
deadline); negative / non-finite / malformed override + malformed VT fail-fast;
named helpers replace the lambda idiom; float asserts use pytest.approx (Sonar
S1244). 43 tests pass; e2e re-verified.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
67f7d6e to
20462e0
Compare
|
| Filename | Overview |
|---|---|
| workers/pg_queue_consumer/supervisor.py | Core fix: adds shutdown_grace_from_env() (VT-tracking, validated, floored at 30s) and converts _join_children to a single shared deadline with an at-least-once poll guard in _wait_for_exit; logic is sound and well-guarded. |
| workers/tests/test_pg_consumer_supervisor.py | Adds 9 new TestShutdownGraceFromEnv cases (VT tracking, floor, override, validation, malformed inputs) and 2 TestJoinChildren regression tests verifying the shared-deadline property; coverage is thorough. |
| workers/queue_backend/pg_queue/README.md | New reference doc listing all env vars, glossary terms, and table schemas for the PG queue; purely documentation with no functional impact. |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant k8s as k8s/HPA
participant sup as Supervisor
participant c1 as Child 1 (in-flight)
participant c2 as Child 2 (idle)
Note over sup: startup: grace = max(30, VT) e.g. 9060s
k8s->>sup: SIGTERM
sup->>sup: _on_term: stopping.set()
sup->>c1: SIGTERM at T0
sup->>c2: SIGTERM at T0
c2-->>sup: exits quickly
Note over sup: finally block runs
sup->>c1: SIGTERM again (harmless duplicate)
sup->>c2: SIGTERM again (ProcessLookupError suppressed)
Note over sup: _join_children: deadline = now + grace_seconds (ONE shared value)
sup->>c1: _wait_for_exit(pid1, deadline) polls every 0.1s
c1-->>sup: exits within grace period
sup->>c2: _wait_for_exit(pid2, deadline) polls at least once
Note over c2: already exited, first poll returns True
Note over sup: all children drained, supervisor exits cleanly
%%{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 k8s as k8s/HPA
participant sup as Supervisor
participant c1 as Child 1 (in-flight)
participant c2 as Child 2 (idle)
Note over sup: startup: grace = max(30, VT) e.g. 9060s
k8s->>sup: SIGTERM
sup->>sup: _on_term: stopping.set()
sup->>c1: SIGTERM at T0
sup->>c2: SIGTERM at T0
c2-->>sup: exits quickly
Note over sup: finally block runs
sup->>c1: SIGTERM again (harmless duplicate)
sup->>c2: SIGTERM again (ProcessLookupError suppressed)
Note over sup: _join_children: deadline = now + grace_seconds (ONE shared value)
sup->>c1: _wait_for_exit(pid1, deadline) polls every 0.1s
c1-->>sup: exits within grace period
sup->>c2: _wait_for_exit(pid2, deadline) polls at least once
Note over c2: already exited, first poll returns True
Note over sup: all children drained, supervisor exits cleanly
Reviews (2): Last reviewed commit: "UN-3695 [FIX] SIGKILL warning log uses %..." | Re-trigger Greptile
|
Addressed the Greptile nit ( |
b38f8b3 to
56386bc
Compare
… (Greptile nit) The _join_children SIGKILL warning printed grace_seconds with %s (e.g. 9060.0s) while the startup log uses %.0f (9060s). Align to %.0fs. Cosmetic — no behaviour. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
56386bc to
77e0f52
Compare
5a385a4
into
feat/UN-3445-pg-queue-integration
|



What
The PG-queue prefork supervisor SIGKILLed its own children after a hardcoded 30s on shutdown (
_SHUTDOWN_GRACE_SECONDS = 30.0), even though the pod'sterminationGracePeriodSecondsis set toVT + buffer(9120s forworker-pg-file-processing, VT=9060). So on a graceful SIGTERM (deploy / HPA pod scale-down) the supervisor knifed an in-flight batch ~300× sooner than the grace the pod actually had — orphaning the batch: the executor finished the work, but nothing was left to consume the reply, mark the file COMPLETE, decrement the barrier, or ack the queue message.Two changes:
shutdown_grace_from_env()— the per-child drain grace now defaults to the consumer VT (WORKER_PG_QUEUE_CONSUMER_VT_SECONDS), floored at 30 so it can never be shorter than before (non-regressive: OSS/dev on the default VT stays 30; fileproc gets 9060 with no chart change, since the chart already sets VT). ExplicitWORKER_PG_QUEUE_CONSUMER_SHUTDOWN_GRACE_SECONDSoverrides._join_children→ single shared deadline (was per-child recomputed). Children are SIGTERM'd together and drain in parallel, so one shared window still gives each its full grace while bounding total drain to ~grace. A per-child deadline sums to N×grace and, at grace=VT, blows past the pod grace → k8s SIGKILLs the whole pod, hard-killing siblings that were draining cleanly.Why
Deploys and autoscaling routinely send graceful SIGTERM to workers. The 30s app-grace silently defeated the ~2.5h drain budget the chart deliberately provisions, so every scale-down/rollout landing mid-extraction stranded a batch until visibility-timeout redelivery (~2.5h) or the reaper.
Scope / safety
pg_queue_consumer) is a PG-worker-only component — it never runs for Celery workers, so no feature-flag branch is needed and the Celery path is byte-for-byte unchanged.max(30, VT)floor guarantees the drain is never shorter than today.How tested
shutdown_grace_from_env(VT-tracking / floor / override) + a shared-deadline regression test — 38 supervisor tests pass;test_pg_consumer_supervisor.py+test_pg_queue_consumer.py= 109 pass.Part of UN-3695 (PG crash-safe worker shutdown). Follow-ups tracked there: heartbeat-lease for hard-kill recovery (next); the idempotency track was found already handled (billing is idempotent in the cloud plugins) and needs no code.
🤖 Generated with Claude Code