test

---## Code Review### Overview

Code Review

Overview

Config-only fix that raises the consumer-side librdkafka fetch ceilings to match the 64 MiB max.message.bytes set on the producer/broker side. The root cause analysis is correct — the mismatch between broker-stored message size and consumer fetch ceiling is exactly what causes the MSG_SIZE_TOO_LARGE re-seek loop described in the PR body.

The fix is correct ✓

• max.partition.fetch.bytes = 64 MiB: per-partition ceiling, must be ≥ max.message.bytes
• fetch.max.bytes = 64 MiB: total per-fetch ceiling, matching the partition limit
• receive.message.max.bytes = 100 MiB: socket buffer ≥ fetch.max.bytes + overhead, satisfies librdkafka's internal validation

These exactly mirror the producer settings already on lines 118–119, which is the right invariant to enforce.

Suggestions

1. Over-verbose block comment (succinctness)

The 5-line comment block (lines 473–477) mostly reproduces the PR description. The existing file style uses terse inline comments (see -- 1 MiB batches, -- 1 GiB librdkafka queue cap). Terse inline comments would match better and save 4 lines:

<> K.extraProp "max.partition.fetch.bytes" "67108864"  -- 64 MiB, matches broker max.message.bytes
<> K.extraProp "fetch.max.bytes"           "67108864"  -- 64 MiB total per fetch
<> K.extraProp "receive.message.max.bytes" "104857600" -- 100 MiB socket buffer (headroom over fetch.max.bytes)

2. Duplicated magic numbers

67108864 now appears on lines 118, 478, and 479; 104857600 on lines 119 and 480. The producer block (lines 118–119) has the same issue and no inline human-readable label at all. Centralising as top-level constants (or at minimum adding inline size labels) would prevent future producer/consumer drift — the very problem this PR fixes:

kafkaMaxMessageBytes, kafkaReceiveBufferBytes :: String
kafkaMaxMessageBytes    = "67108864"  -- 64 MiB
kafkaReceiveBufferBytes = "104857600" -- 100 MiB

3. Minor: fetch.max.bytes = max.partition.fetch.bytes

Setting them equal means a single 64 MiB message saturates the entire fetch response, blocking simultaneous fetches from other assigned partitions. Likely intentional for the DLQ case, but worth a note if this consumer group handles many partitions.

Summary

The fix is correct and addresses an observed production issue. The main feedback is style: the block comment is verbose relative to the rest of the file, and the duplicated magic-number strings are now a three-way sync point. Neither is a blocker.

(Deleting test comments — this is the real review below.)

Code Review

Overview

Two independent changes: (1) a config-only fix raising Kafka consumer fetch ceilings to match the 64 MiB broker max.message.bytes, and (2) a new regression-test file for cancelAllConcurrently. Both are correct. A few succinctness and style notes below.

src/Pkg/Queue.hs — consumer fetch ceiling

Fix is correct. The three props (max.partition.fetch.bytes, fetch.max.bytes, receive.message.max.bytes) are the right knobs and the values match the producer config at lines 118–119. No issues with the logic.

Magic number duplication. "67108864" now appears three times in this file (lines 118, 478, 479) and "104857600" twice (lines 119, 480). These are config string literals so a top-level let or where binding doesn't help much, but even a module-level constant avoids the silent divergence risk:

kafkaMaxMessageBytes, kafkaReceiveMaxBytes :: Text
kafkaMaxMessageBytes  = "67108864"   -- 64 MiB
kafkaReceiveMaxBytes  = "104857600"  -- 100 MiB (= 64 MiB + librdkafka overhead)

Both producer and consumer props would then reference these, making a future size change a single-site edit.

Comment verbosity. The 4-line block comment (lines 473–477) largely duplicates the PR description; the existing inline comments in this function are all one-liners. Could be trimmed to:

-- Consumer fetch ceiling must match producer/broker 64 MiB max.message.bytes;
-- defaults (1 MiB partition, 50 MiB total) cause MSG_SIZE_TOO_LARGE → re-seek loop.

test/unit/System/ServerSpec.hs — cancelAllConcurrently regression tests

Control.Exception vs UnliftIO.Exception — intentionally correct. The slow fiber deliberately swallows AsyncCancelled via catch \(_ :: SomeException) -> threadDelay 2_000_000. UnliftIO.Exception.catch re-throws async exceptions, which would break the test. Using base Control.Exception here is the right call; worth a short comment so a future reader doesn't "fix" it:

-- Intentionally uses Control.Exception.catch (not UnliftIO) to swallow AsyncCancelled
slow <- async $ forever (threadDelay 10_000) `catch` \(_ :: SomeException) -> threadDelay 2_000_000

Date/bug-ref in module comment (line 11). The -- Regression for the 2026-06-22 "Ctrl-C …" bug: comment will rot — the date means nothing to a future reader and the bug context belongs in the commit message / PR body. Drop the date and inline prose; the test names already tell the story:

-- Regression suite for cancelAllConcurrently; sequential cancel blocked on slow fibers.

Test timing (400 ms delay, test 1). The 400 ms is conservative but fine in practice; just note that slow's 2 s cleanup means the full test run takes ~2 s. Not a bug, but worth knowing for CI budget.

Test 2 race assertion. shouldReturn Right () is clean; no issues.

Summary

Code Review

Overview

A targeted, well-motivated fix for a real production incident. The DLQ replay group wedging is clearly explained and the config values are consistent with librdkafka constraints. The test file additions are separate unrelated work (new regression tests for cancelAllConcurrently).

src/Pkg/Queue.hs — Kafka consumer fetch ceiling

Magic numbers duplicated across producer and consumer

67108864 and 104857600 already appear verbatim in the producer config at lines 118–119 (message.max.bytes / receive.message.max.bytes). Now they're repeated in consumerProps. If one side gets bumped, the other silently diverges.

These are string-typed (extraProp :: Text -> Text -> …) so constants can't be shared directly without a show, but a short helper avoids the coupling:

-- module-level
mib :: Int -> Text
mib n = show (n * 1024 * 1024)

then:

<> K.extraProp "max.partition.fetch.bytes" (mib 64)
<> K.extraProp "fetch.max.bytes"           (mib 64)
<> K.extraProp "receive.message.max.bytes" (mib 100)

(Same helper for the producer.) That also makes the values self-documenting without inline comments.

Comment block

The two-line -- Fetch ceiling must match… comment is a valid WHY, and the inline -- 64 MiB tags are useful. Fine as-is if you keep the raw numbers; redundant if you switch to mib.

test/unit/System/ServerSpec.hs — New imports

Control.Exception is inconsistent with project conventions

The codebase overwhelmingly uses UnliftIO.Exception for catch / finally:

• src/BackgroundJobs.hs:112 — UnliftIO.Exception (bracket, catch, throwIO, try, tryAny)
• src/System/Tracing.hs:36 — UnliftIO.Exception (bracket, finally, withException)
• src/CLI/Core.hs:59 — UnliftIO.Exception (catch)
• src/CLI/Commands.hs:73 — UnliftIO.Exception (catch, tryAny)

The test runs in plain IO so both are equivalent, but importing from Control.Exception breaks the pattern. Swap to UnliftIO.Exception.

catch / finally may already be in scope via Relude

Relude re-exports catch and finally from Control.Exception, so this import may be redundant. If UnliftIO.Exception is used instead (which Relude does not re-export), the explicit import is needed and correct.

Timing in the first test

threadDelay 400_000 (400 ms) as the assertion window is generous for a concurrent-cancel operation that should resolve in microseconds, so it won't be a flakiness source. Just noting it was checked.

Summary

The fix itself is sound and should be landed promptly per the deploy sequencing in the PR description. The style items above are low-risk but worth cleaning up in a follow-up or before merge.