fix: concurrent bounded shutdown so Ctrl-C/SIGTERM actually exits#440
Conversation
Replay byte-batches the due prefix (grouped by ce-type + write target) instead of one bulk write per record, so a backlog drains in ~64 MiB writes like primary ingest. Replay also rewrites only the originally-failed store: a WriteTarget (WriteBoth/WritePgOnly/WriteTfOnly) derived from the monoscope-write-failure header threads through processList -> insertAndHandOff -> bulkInsertOtelLogsAndSpansTF (new TF-only branch). Avoids duplicating the still-durable leg (PG inserts aren't idempotent, so a tf-failed replay would otherwise double the PG row). Live ingest paths pass WriteBoth.
Integration (TimefusionWriteFailureSpec): tf-failed replay writes exactly one row — pins the no-duplicate-PG-row fix; pg-failed replay skips a down TimeFusion leg (Right where a normal msg DLQs). Pure header->target (writeTargetFor) and DLQ-header (writeFailureDlqHeaders) checks moved from integration `it`s to doctests.
runServer ended in async's waitAnyCancel, whose `mapM_ cancel` cancels fibers sequentially and blocks on each one's death before signalling the next. One slow-to-die fiber (Warp's graceful drain, a wedged librdkafka poll) kept every later fiber alive — so Ctrl-C left the kafka workers retrying dead-DB writes and emitting consumer metrics instead of exiting. Replace with awaitShutdown: install SIGINT+SIGTERM handlers, race a signal against any fiber exiting, then cancelAllConcurrently — mapConcurrently_ cancel under a 15s deadline. Teardown is now bounded by the slowest single fiber, not the sum, and a wedged fiber can't hang the process. Catching SIGTERM also lets docker stop / k8s drain gracefully instead of SIGKILL. cancelAllConcurrently is exported for the regression test in ServerSpec.
Code ReviewOverall this is a clean, well-motivated PR. The core shutdown fix is correct and elegant, and the
|
| Area | Severity |
|---|---|
writeTargetFor False "tf-failed" → attempts disabled store |
Worth a logAttention / doc note |
Map.fromListWith (<>) reversal fixed downstream by sort |
Minor — use flip (<>) |
slow/stuck asyncs leak in unit tests |
Minor — use withAsync |
singleLeg losts lambda readability |
Nit |
cancelAllConcurrently export |
Nit |
The shutdown fix and WriteTarget refactor are both correct and improve the codebase. Happy to approve once the fromListWith flip and withAsync nits are considered.
…break) Every "Build and Test" run has failed — across all PRs and the merged ones — compiling test/unit/System/ServerSpec.hs (added in #440) with: GHC-87110: Could not load module 'Control.Concurrent'. It is a member of the hidden package 'base-4.21.0.0'. Perhaps you need to add 'base' to build-depends. ServerSpec imports Control.Concurrent / Control.Exception directly from base; every other test/unit module goes through relude, so the unit-tests suite never listed base. With NoImplicitPrelude the direct base imports can't resolve and the suite fails to compile (the varying error line across runs was just whichever import GHC reached first — not a flake). Add base to the unit-tests build-depends.
…break) Every "Build and Test" run has failed — across all PRs and the merged ones — compiling test/unit/System/ServerSpec.hs (added in #440) with: GHC-87110: Could not load module 'Control.Concurrent'. It is a member of the hidden package 'base-4.21.0.0'. Perhaps you need to add 'base' to build-depends. ServerSpec imports Control.Concurrent / Control.Exception directly from base; every other test/unit module goes through relude, so the unit-tests suite never listed base. With NoImplicitPrelude the direct base imports can't resolve and the suite fails to compile (the varying error line across runs was just whichever import GHC reached first — not a flake). Add base to the unit-tests build-depends.
* fix(queue): self-heal DLQ commit-starvation via reseek on no-ack A consumer chunk that returns no acks (transient write/DLQ-publish failure → routeBatchOutcome []) does not advance the commit base, but the poll position has already moved past it. librdkafka never redelivers uncommitted offsets in-session, so the base pins forever while the consumer forward-processes to log-end and goes idle — the 2026-06-25 apitoolkit_eu_dlq wedge (~7.6M lag, committed offsets frozen). Fix: on no-ack, the worker records the chunk's lowest offset in reseekVar; the poll thread (which owns all Kafka ops) seeks each such partition back to that offset and resets its tracker so the un-acked chunk is redelivered and retried once the failure clears. Seeks backward only (never past the committed base) so it can never skip — worst case redelivers already-processed offsets, which is idempotent downstream. Regression test in KafkaConsumerSpec drives a chunk that no-acks on first processing then acks: without the fix the base pins and the watermark never recovers (drive times out); with it the reseek redelivers and the commit advances. Adds SeekPartitions to the in-memory broker mock. * fix(queue): address reseek review — prune reseekVar on rebalance + test rigor - Prune reseekVar to the live partition set in the committer (alongside tracker/committed). Without it, a partition that stalled (reseek recorded) then got revoked would make the poll thread seek a partition this consumer no longer owns and re-seed a stale tracker entry the prune can never reach. - Tracker reset on reseek uses the left-biased Map union (Map.map (`PartProgress` mempty) reseeks <> tk) instead of folding a toList. - Test: assert the retry actually fired (attemptVar >= 2), so removing the no-ack guard can't make the test pass trivially. - Test mock: key the broker cursor by (topic, partition) so same-topic multi-partition seeks no longer silently collide. (head-instead-of-minimum was not applied: head trips -Werror=x-partial; minimum is the lint-clean partial reducer and the offset list is small.) * fix(ci): add base to unit-tests build-depends (fixes GHC-87110 build break) Every "Build and Test" run has failed — across all PRs and the merged ones — compiling test/unit/System/ServerSpec.hs (added in #440) with: GHC-87110: Could not load module 'Control.Concurrent'. It is a member of the hidden package 'base-4.21.0.0'. Perhaps you need to add 'base' to build-depends. ServerSpec imports Control.Concurrent / Control.Exception directly from base; every other test/unit module goes through relude, so the unit-tests suite never listed base. With NoImplicitPrelude the direct base imports can't resolve and the suite fails to compile (the varying error line across runs was just whichever import GHC reached first — not a flake). Add base to the unit-tests build-depends.
…break) (#444) Every "Build and Test" run has failed — across all PRs and the merged ones — compiling test/unit/System/ServerSpec.hs (added in #440) with: GHC-87110: Could not load module 'Control.Concurrent'. It is a member of the hidden package 'base-4.21.0.0'. Perhaps you need to add 'base' to build-depends. ServerSpec imports Control.Concurrent / Control.Exception directly from base; every other test/unit module goes through relude, so the unit-tests suite never listed base. With NoImplicitPrelude the direct base imports can't resolve and the suite fails to compile (the varying error line across runs was just whichever import GHC reached first — not a flake). Add base to the unit-tests build-depends.
1 participant
Problem
Hitting Ctrl-C didn't shut the app down — the kafka workers kept retrying dead-DB writes (
retryHasqlWritetransient loop) andkafka.consumer.metricskept emitting, instead of the process exiting.Root cause
runServerended in async'swaitAnyCancel asyncs:mapM_ cancelcancels fibers sequentially, blocking on each one's death before signalling the next. So one slow-to-die fiber early in the list (Warp's 30s graceful drain, a wedged librdkafka poll, a long DB op) keeps every later fiber alive — including the kafka workers, whosecancelwas queued behind it.(Note: this is not
retryHasqlWriteswallowing the async exception — it usesUnliftIO.tryAny, which rethrows async exceptions. The cancel simply never reached the retry loop.)Fix
Replace
waitAnyCancelwithawaitShutdown:shutdownMonoscope(drain/flush/close pools) and exit.docker stop/ k8s now drain gracefully instead of SIGKILL-dropping in-flight buffers.cancelAllConcurrentlyis exported solely for the regression test.Test
test/unit/System/ServerSpec.hsproves a slow-to-die fiber listed first no longer blocks cancelling the others, and that the deadline is honored when a fiber never dies. Addedasyncto theunit-testsdeps.System.Serveritself compiles cleanly (ghcid "All good").