Bound the engine drain so the shutdown after it actually runs (#192)

[richardmhope]

Closes #192.

The defect

dramatiq.Worker.stop() waits ten minutes by default. Both grace periods this project ships are two (stop_grace_period: 120s, terminationGracePeriodSeconds: 120). So an engine holding a long fetch at SIGTERM was still inside that wait when SIGKILL arrived, and none of the shutdown after it ran: no heartbeat.stop(), no close_api_client(), no server.shutdown(), no engine_stopped in the log.

Three sets of comments — compose, the chart, and the runner's except Interrupt handler — described a drain that did not happen.

The policy decision

The issue suggested adding notify_shutdown: True so the runner's existing Interrupt path reports partial work, on the assumption that "checkpointed connectors resume cleanly". They don't, on that path, and I've decided the other way:

REPORTABLE_STATUSES = (COMPLETED, FAILED) — an engine has no third thing to say. reclaim_expired_leases only reclaims tasks in LEASED/RUNNING, so a task that reports failed is terminal: never redelivered, and it makes its scan partial, which may not auto-resolve findings (ADR 0009 §4).

So interrupting would trade one lease TTL of latency for a scan that cannot close a secret somebody has already fixed — on every rolling deploy that lands mid-scan. Waiting and then letting the lease lapse costs latency instead, and keeps API reclaim the single re-delivery authority (ADR 0009 §2). A checkpointing connector genuinely does resume from its last flushed batch on that path.

The policy is now written down in docs/deployment.md § Draining an engine, including the rejected alternative, so the next person doesn't have to re-derive it.

Considered and not done here: an explicit POST /scan-tasks/{id}/release would give both — no lease wait and no terminal failure. It also adds a second path back onto the queue, which is exactly what ADR 0009 §2 forbids in spirit, so it's an ADR-level decision rather than something to slip into a drain fix. Happy to raise it as a follow-up if you want the latency back.

The change

• ICEBERG_DRAIN_SECONDS (default 90) bounds the wait; drain() in worker.py does the waiting and names what it gave up on.
• A task still running at the deadline produces an engine_drain_incomplete warning listing the task ids — the scans about to sit on a lease.
• Comments in compose, the chart and runner.py now say what actually happens. The runner's says why a shutdown does not reach its Interrupt handler.

Tests

• tests/test_deploy_invariants.py — the default budget is strictly below both grace periods. This is the property that makes the shutdown reachable at all, so it belongs with the other deploy invariants rather than in prose.
• apps/engine/tests/test_worker.py — a real Worker with a task in flight: the drain returns on a 0.25s budget and reports the task as abandoned. Verified to fail against the unbounded call (it waits out the fixture and reports nothing abandoned). A second test pins that the budget is a ceiling, not a delay — an idle drain returns immediately.

make check green: ruff, mypy, docs check, 1956 passed / 2 skipped.

Operator impact

None required. Raising ICEBERG_DRAIN_SECONDS for routinely-long fetches means raising both grace periods with it, which the doc says explicitly.

---

Generated by Claude Code

[icebergai-review-bot]

Verdict

APPROVE

Completed bounded review across 1 immutable scope(s). One high-severity shutdown regression found.

Scope health

Convergence: healthy. Review mode: initial.
Recommended action: CONTINUE_INCREMENTAL.

• No escalation signals.

Prior findings

Finding	Status
—	No prior finding state

New findings

Root cause: A per-thread join timeout is treated as a total shutdown deadline.

• FYI · medium: Drain timeout is applied per worker thread, not to the overall drain — apps/engine/src/iceberg_engine/worker.py
  Status: NEW. Attribution: new_in_scope.
  dramatiq.Worker.stop(timeout=...) joins each consumer/worker thread with the supplied timeout. Passing the full drain budget once therefore permits multiple concurrently running actors to consume successive full budgets. With the shipped four worker threads and a 90s budget, two blocking actors can keep shutdown inside consumer.stop() for about 180s, exceeding the 120s Docker/Kubernetes grace period.
  Invariant: The total drain wait must finish before every shipped 120-second termination grace period.
  Ownership: Engine shutdown / Dramatiq worker lifecycle. Behaviour: Rolling deploys and container shutdowns with more than one long-running task.
  Evidence: drain() calls consumer.stop(timeout=int(seconds * 1000)) once. Dramatiq applies that timeout to each thread join; the new test covers only one running actor, so it does not exercise the cumulative wait. Reproduce by running two actors that block longer than 90s on the default four-thread worker, invoke drain(consumer, 90), and observe that cleanup has not run by the 120s termination grace.
  Independent assessment: Downgraded to advisory: The pinned Dramatiq 2.2.0 implementation passes worker threads to join_all(), which deducts elapsed time after each join; the 90-second timeout is a total budget for the worker-thread group, not a separate 90-second wait per actor. The proposed cumulative 180-second behavior is therefore not established.

Fix-induced regressions

• None evidenced.

Uncertainty

• No material uncertainty recorded.

Validation

• Reviewed the supplied immutable diff only.
• Exact-head CI was reported as passed.

Residual risks

• None identified.