Code Review — PR #437

Overall this is a well-structured refactor. The effect-abstraction is clean, the flush-before-commit durability gate is the right fix, and the pure-function extraction with doctests is exactly the right approach. Notes below, prioritised by impact.

Potential Bug — takeWhile isDue ordering assumption (medium risk)

In subChunksFor KafkaDlqReplay, takeWhile (isDue nowEpoch) assumes monoscope-next-due-at is monotonically non-decreasing with offset within a topic-partition. That holds when processing latency ≪ tier delay, but under a slow DB call a late-published message can block a partition's ready tail. The PR description notes "offset = failure-time = due-time order" — this invariant deserves a comment in the code so future readers know the assumption.

Catch-all error in interpreters

_ -> error "runSharedKafkaProducer: unsupported KafkaProducer operation"

Same in runMockProducer / runMockConsumer. If kafka-effectful adds a constructor, this panics silently. If the GADT is closed, remove the wildcard and let -Wincomplete-patterns catch additions at compile time. If it's open, add a comment explaining why the fallback is forced.

Conciseness / package usage

BC.pack . toString on Text is a two-step conversion; encodeUtf8 is already in scope via Relude:

-- current (publishRawToKafka)
map (\(k, v) -> (BC.pack $ toString k, BC.pack $ toString v)) (HM.toList attributes)
-- simpler
bimap encodeUtf8 encodeUtf8 <$> HM.toList attributes

isDue does decodeUtf8 @Text . snd then toString before readMaybe — a ByteString → Text → String round-trip. BC.unpack . snd is direct and BC is already imported.

Map.insertWith (\_ old -> old) → Map.insertWith (flip const) — flip const is the idiomatic name for "keep old on collision".

when (not (null commits) || inflight > 0) → unless (null commits && inflight == 0) — one negation instead of two.

The 5-tuple returned by subChunksFor:

[((Text, K.PartitionId), ConsumerRecord ..., [(Text, ByteString)], [Int64], Int)]

A named record or type alias would make the worker pattern-match self-documenting. This is the biggest readability drag in the PR.

The committer's tuple-section (cs,,) . Map.size <$> readTVar trackerVar <*> readTVar inflightVar is correct but non-obvious. Explicit bindings inside atomically do are clearer and not longer.

Minor observations

• dlqHeaders moved to a where clause hides it from direct unit test import; a module-level private binding is easier to test in isolation.
• minimum offs is safe by chunksByBytes' non-empty-chunk invariant, but NonEmpty for offs would make that structural.
• consumerRecordHeadersToHashMap / ceTypeFor / topicToCeType are pure self-contained helpers living in where clauses — module-level private functions cost nothing and are directly testable.
• Server.hs: map fst (Queue.retryTiers ...) for the DLQ subscription set is cleaner than a hard-coded list — good change.

Summary

Architecture is sound, the in-memory broker test that exercises the flush-before-commit contract is exactly the right design, and the watermark committer tests cover the data-loss-critical invariant well. Main asks: (a) document or harden the takeWhile isDue ordering assumption, (b) replace catch-all error with narrower matches or a comment explaining why it's unavoidable, (c) apply the bimap encodeUtf8 / flip const / BC.unpack simplifications before merge.

Code Review — kafka: unified Eff-based consumer + producer, tiered DLQ, flush-gated commits

Overall this is a solid refactor: the effectful consumer/producer abstraction tests beautifully, the watermark committer invariants are well-doctest'd, and the Ki-scoped three-thread decomposition is clean. A few issues worth addressing before merging.

Bug / Behavioral Regression

fromRight [] silently swallows KafkaErrors in pubsubService (src/Pkg/Queue.hs:230)

$ fromRight []
<$> runBackground … (runErrorNoCallStack @K.KafkaError (runSharedKafkaProducer appCtx (routeBatchOutcome …)))

A Left KafkaError becomes [] (empty ack list), which then passes to the Google PubSub acknowledgeRequest — acking messages whose DLQ publish silently failed. The old code returned IO [Text] directly and the caller used the result directly; there was no swallowed error. This is a regression: a DLQ failure in the pubsub path now looks like a successful empty batch. Either propagate the error or log-and-hold (return [] to block the ack).

Correctness Concerns

IntSet storing Int64 offsets via fromIntegral (completeOffsets, line ~373)

ins s o = if o >= base0 then IntSet.insert (fromIntegral o) s else s

IntSet stores Int; on a 64-bit host Int is 64-bit so this is fine in practice, but the truncation is implicit. Using Data.IntMap.Strict keyed on Int (or an explicit Map Int64) would make the assumption visible. Data.Map.Strict is already imported; a Map Int64 () would be type-safe with no performance penalty for the sizes involved here, and the advance loop would read cleaner.

DLQ backoff only fires when the entire poll batch is not-yet-due (decoupledLoop, line ~563)

when (role == KafkaDlqReplay && null subChunks && not (null rs)) (threadDelay 5_000_000)

If a batch mixes due and not-yet-due records, subChunks is non-null (due records) and the backoff is skipped. The not-yet-due records stay uncommitted and are redelivered on the next 100ms poll indefinitely until all due records drain. In a typical retry carousel this means continuous tight re-polling of mixed batches. The condition should check whether any not-yet-due records remain, not whether subChunks is empty.

Runtime panic for future KafkaProducer operations (runSharedKafkaProducer, line ~144)

_ -> error "runSharedKafkaProducer: unsupported KafkaProducer operation"

Same pattern in test mocks. If kafka-effectful adds a new operation (e.g. BeginTransaction), callers hit a runtime panic instead of a compile error. A non-exhaustive interpret is acceptable if the library guarantees the set is closed, but it should be documented. At minimum, add a note that this must be revisited when upgrading kafka-effectful.

Code Succinctness / Package Usage

Header conversion — use bimap (publishRawToKafka, line ~351)

-- current
map (\(k, v) -> (BC.pack $ toString k, BC.pack $ toString v)) $ HM.toList attributes
-- shorter with bimap (Data.Bifunctor, already in scope via Relude)
map (bimap (BC.pack . toString) (BC.pack . toString)) $ HM.toList attributes

unless reads better than when (not …) (committer metrics, line ~536)

-- current
when (not (null commits) || inflight > 0) $ …
-- idiomatic
unless (null commits && inflight == 0) $ …

"" <$ result is a code smell (Pages/Replay.hs:159)

pure $ first ("Failed to publish replay event: " <>) ("" <$ result)

publishJSONToKafka now returns Eff es (), so runSharedProducer already yields Either Text (). The "" <$ exists only to satisfy a caller that still expects Either Text Text. The return type of publishReplayEvent (or the chain leading to it) should be updated to Either Text () to eliminate this.

dlqHeaders moved into where — doctest coverage

The function had a doctest when it was top-level. It still has the doctest block in the where clause, but most doctest runners (including doctest-parallel) only discover top-level bindings unless you configure --no-magic. Verify the doctest suite still runs those examples; if not, promote the examples to the parent function's haddock or make dlqHeaders a top-level internal (unexported) function.

Minor

• subChunksFor lacks doctests despite being the most complex pure function in the file and being explicitly called out in the PR description as a decision function. The byte-chunking and isDue filter are both worth doctest coverage to match the bar set by chunksByBytes/completeOffsets.

• Map.insertWith (\_ old -> old) (tracker init in pollOnce) — Map.insertWith const is the conventional spelling of "keep existing value on collision" and is shorter.

• The retryTiers base first element has a 5s delay ((base, 5)), meaning first-failure messages now wait 5s before replay. The old DLQ had no per-message delay (only a 30s global tail-backoff). This is a deliberate change (prevent thrashing) but not mentioned in the PR description; worth a changelog note for ops.

Thanks for the thorough review. Disposition below.

Implemented (1292040)

• bimap in publishRawToKafka — done.
• unless (null commits && inflight == 0) in the committer metrics — done.
• Non-exhaustive runSharedKafkaProducer interpret — added a comment documenting the closed-set assumption and the revisit-on-kafka-effectful-upgrade note (also points at the mock).
• subChunksFor doctest — added one covering the DLQ due-prefix filter (the non-obvious branch; the primary branch is just chunksByBytes, already doctested).
• "" <$ result in Replay.hs — publishReplayEvent now returns Either Text () and the response drops the always-empty messageId.

Declined, with reasoning

• Map.insertWith const — this inverts the semantics. insertWith f calls f new old on collision; const new old = new, so it would overwrite the partition's committed base with each poll's new minimum offset, jumping the watermark forward past uncommitted offsets — a data-loss bug. \_ old -> old (keep existing base) is correct; kept as-is.
• IntSet → Map Int64 () — Int is 64-bit on our hosts, so IntSet holds the offsets losslessly, and its PATRICIA representation is more compact and faster than a boxed-key Map Int64 (). Switching would add allocation, which cuts against the minimal-allocation goal of this path. Kept IntSet.
• fromRight [] "acks failed-DLQ messages" — it doesn't: the ack is guarded by unless (null msgIds), so [] means no ack → redelivery (the safe outcome). And routeBatchOutcome catches KafkaError internally via tryError and logs each DLQ branch, so the error never escapes to the runErrorNoCallStack boundary in the first place — fromRight [] is unreachable-defensive, not a swallow. Left as-is.
• dlqHeaders where-clause doctest — this repo uses the classic Test.DocTest runner, which extracts examples from all comments (including where bindings), so those examples do run; the doctest-parallel --no-magic caveat doesn't apply here.

Follow-up (out of scope for this PR)

• DLQ mixed-batch backoff — correct that a mixed due/not-due batch skips the 5s nap. But the underlying behavior is subtler than the nap heuristic: pollMessageBatch advances the in-session position past the not-yet-due records (we don't seek), so they aren't re-fetched until a rebalance/restart resets to the committed base. Tightening this correctly is a test-backed change to the DLQ-replay seek/commit semantics — pre-existing, not introduced by this PR — so I'd rather do it separately than tweak the nap condition here.

Noted

• First-tier 5s delay — added an ops note to the PR description.

The prior two reviews were thorough and 1292040 addressed most actionable points. A few items not yet surfaced:

isDue: unnecessary Text intermediate

decodeUtf8 @Text . snd then toString is a ByteString → Text → String round-trip. BC is already imported:

-- current
readMaybe . toString . decodeUtf8 @Text . snd =<< find …
-- simpler – BC is already in scope
readMaybe . BC.unpack . snd =<< find …

Map.insertWith (flip const) was never addressed

The second review incorrectly suggested const (which overwrites — author rightly declined). But the first review suggested flip const, which is correct: flip const new old = old. This is the standard idiom for "keep existing value on collision" and shorter than the lambda. It was never acknowledged in the author's response.

subChunksFor 5-tuple: concrete fix

Both reviews flagged this; neither proposed one. A type alias costs zero runtime and fixes the positional-counting:

type WorkItem =
  ( (Text, K.PartitionId)
  , K.ConsumerRecord (Maybe ByteString) (Maybe ByteString)
  , [(Text, ByteString)]  -- payloads
  , [Int64]               -- offsets
  , Int                   -- bytes
  )

The worker pattern-match and the for_ in pollOnce would then be legible without counting commas.

Migration file bundled without context

static/migrations/0100_db_response_status_code_to_text.sql is added to extra-source-files in the cabal file, but its content is not in the diff and the PR description doesn't mention it. Intentional inclusion?

Still open from the first review

• takeWhile isDue ordering assumption — add a code comment stating that monoscope-next-due-at is monotonically non-decreasing with offset within a partition (by the invariant that offset ≈ failure-time ≈ due-time), so takeWhile is safe. Not addressed in 1292040 or the author's response.
• Committer's (cs,,) tuple-section — still harder to read than three explicit let/<- bindings inside atomically do; mentioned in the first review, not addressed.

Code Review — PR #437: Unified Eff-based Kafka consumer + producer, tiered DLQ, flush-gated commits

Overall: This is a well-structured rewrite. The effect-based abstraction, pure decision logic with doctests, and flush-before-commit durability gate are all solid design choices. A few specific issues below, roughly in priority order.

Bugs / correctness

dlqHeaders doctests silently dropped

The function moved from a top-level binding (where doctest finds it) into a where clause inside publishToDeadLetterQueue. Doctest doesn't extract tests from where-clause definitions, so those four examples no longer run. Either keep dlqHeaders top-level (re-export it if only tests need it) or move the examples onto publishToDeadLetterQueue itself.

minimum offs hidden partial

In pollOnce:

modifyTVar' trackerVar (Map.insertWith (flip const) tpKey (PartProgress (minimum offs) mempty))

minimum from Data.List.Extra still throws on empty. The invariant (offs is always non-empty from subChunksFor) holds today, but it's fragile. viaNonEmpty minimum offs or a NonEmpty on offsets would encode this at the type level, or at least a fromMaybe 0.

Performance

size x called twice in chunksByBytes

| n + size x > target = reverse acc : go [x] (size x) xs

When the guard fires, size x is evaluated twice (once in the guard, once as the argument). On a hot path over large batches this adds up. Capture it:

go acc !n (x : xs) =
  let !s = size x
  in if | null acc          -> go [x] s xs
        | n + s > target    -> reverse acc : go [x] s xs
        | otherwise         -> go (x : acc) (n + s) xs

subChunksFor KafkaPrimary iterates each chunk 3×

(tpKey, recc, consumerRecordToTuple <$> chunk, K.unOffset . (.crOffset) <$> chunk, sum (recordBytes <$> chunk))

Three separate fmaps over the same list. One foldl' or unzip3 pass cuts 2/3 of the traversals:

let (tuples, offs, bytes) = foldl' (\(!ts,!os,!b) r -> (ts<>[consumerRecordToTuple r], os<>[K.unOffset r.crOffset], b+recordBytes r)) ([],[],0) chunk

Code quality / succinctness

WorkItem 5-tuple is opaque

type WorkItem =
  ( (Text, K.PartitionId)
  , K.ConsumerRecord (Maybe ByteString) (Maybe ByteString)
  , [(Text, ByteString)]
  , [Int64]
  , Int
  )

A small record (data WorkItem = WorkItem { tpKey :: ..., headerRec :: ..., payloads :: ..., offsets :: ..., totalBytes :: Int }) eliminates the positional destructuring spread through worker, pollOnce, and subChunksFor. The project already uses record syntax heavily, and five-positional destructuring in let (tpKey, recc, tuples, offs, bytes) = ... is fragile against future additions.

IntSet Int/Int64 churn in PartProgress

The explicit fromIntegral round-trips between Int64 offsets and IntSet's Int slots add noise and would silently wrap on a 32-bit host. Since the project imports containers anyway, Set Int64 is drop-in and type-safe:

data PartProgress = PartProgress {base :: !Int64, ahead :: !(Set Int64)}

IntSet is faster, but offset watermark maintenance happens once per second, not in a tight loop — the difference is immeasurable.

committableCommits can use Map.mapWithKey + Map.differenceWith

The current implementation walks Map.toList tracker and rebuilds an advanced association list before folding it back into a map. This can be expressed more directly with the Map-level combinators in containers:

let eligible = Map.mapMaybe (\(PartProgress b _) -> b <$ guard (b > 0)) tracker
    newCommits = Map.differenceWith (\b old -> b <$ guard (b > old)) eligible committed
    tps = [K.TopicPartition (K.TopicName t) p (K.PartitionOffset b) | ((t,p),b) <- Map.toList newCommits]
in (tps, Map.union newCommits committed)

appendRecord in the mock broker is O(n)

let existing = Map.findWithDefault [] topic b.logs
    ...
 in b {logs = Map.insert topic (existing <> [cr]) b.logs}

List append is O(n). For the test sizes (≤5) it's harmless, but swapping to existing ++ [cr] is no better; prepend + reverse at read time, or a Seq, would be idiomatic if the broker mock ever grows.

Runtime safety

Non-exhaustive interpreter panics at runtime on kafka-effectful upgrade

_ -> error "runSharedKafkaProducer: unsupported KafkaProducer operation"

Acknowledged in the comment, but worth flagging: if kafka-effectful adds a new constructor (e.g. transaction support), this is a silent runtime panic rather than a compile-time error. The same wildcard appears in runMockProducer / runMockConsumer in the test. This is an accepted trade-off with effect libraries that use open GADTs, but adding a {-# WARNING ... #-} pragma on kafka-effectful bump commits, or a CI note, would make the blast radius explicit.

Minor

• runSharedProducer wraps errors as toText . show. Kafka errors' Show instance includes the constructor name, which may leak internal names into user-facing API responses on the replay path. Consider a displayException-style helper.
• In pubsubService, the fromRight [] on the outer runBackground result is a valid safety net, but the case where it fires (KafkaError escaping routeBatchOutcome) goes unlogged. A either (\e -> LogBase.logAttention "pubsubService: unexpected KafkaError" ...) pure would surface it.

What's good

• The flush-before-commit gate and its integration test are exactly right — the test validates the regression by design (drop the gate and the test fails).
• Unifying DLQ and primary under one decoupledLoop via KafkaRole is clean; the effect-level swap in tests (runMockProducer / runMockConsumer) gives real confidence in the watermark logic.
• chunksByBytes with doctests, completeOffsets, and retryDestination as pure top-level functions are well-tested and easy to reason about.
• Moving the producer to a lock-free readMVar fast path after init is a nice improvement for write-heavy handlers.

Code Review — kafka: unified Eff-based consumer + producer, tiered DLQ, flush-gated commits

The architecture is solid: decoupled poll/worker/committer threads, pure watermark logic, effect-parameterized consumer, and flush-before-commit gate are all well-designed. The doctests on the pure functions are a good addition. A few concrete suggestions below, prioritised by impact.

🔴 dlqHeaders doctests likely no longer run

dlqHeaders was a top-level binding with verified doctests. Moving it into a where clause inside publishToDeadLetterQueue means most doctest runners (including doctest and doctest-parallel) will silently skip it. The doctests now only serve as documentation. Either move the function back to the top level (exported or not), or duplicate the examples into the parent's haddock.

🔴 Lost observability: DLQ publish failures drop topic context

The old publishToDeadLetterQueue logged each per-message failure with "topic" and "original_topic". The new version just re-throws K.KafkaError and the callers in routeBatchOutcome swallow it silently (Left _ -> pure []). When the DLQ topic is unreachable, operators get no log entry with the affected topic or message origin. Suggest adding at least one Log.logAttention in routeBatchOutcome's Left _ arms before pure [].

🟡 isDue: use lookup instead of find … fst

-- current
find ((== "monoscope-next-due-at") . fst) (K.headersToList r.crHeaders)

-- simpler — lookup is exactly this
lookup "monoscope-next-due-at" (K.headersToList r.crHeaders)

With OverloadedStrings the key literal is already ByteString. Same semantics, one fewer lambda.

🟡 chunksByBytes: merge the null-acc guard

The | null acc branch is only reachable when n == 0 (initial call) and size x > target. Merging it into the preceding guard eliminates the special case:

-- current
go acc !n (x : xs)
  | null acc             = go [x] (size x) xs
  | n + size x > target  = reverse acc : go [x] (size x) xs
  | otherwise            = go (x : acc) (n + size x) xs

-- simpler
go acc !n (x : xs)
  | n + size x > target, not (null acc) = reverse acc : go [x] (size x) xs
  | otherwise                           = go (x : acc) (n + size x) xs

Behaviour is identical: when null acc, n is 0, so n + size x > target is false whenever size x <= target, and the otherwise branch adds x to the empty accumulator; when size x > target and null acc the guard is false so we fall through to otherwise and still emit a singleton chunk.

🟡 committableCommits: use Map.fromList over foldl' Map.insert

-- current
foldl' (\m (tp, b) -> Map.insert tp b m) committed advanced

-- idiomatic — Map (<>) is left-biased; advanced keys win
Map.fromList advanced <> committed

Map.fromList advanced <> committed is one expression, no lambda, and expresses the intent directly (advanced values override committed).

🟡 -- \| typo in dlqHeaders comment

    -- \| Headers stamped on every DLQ publish.

The backslash before | is spurious. Should be -- | (Haddock) or plain --.

🟢 Unrelated change bundled

The otelcol.yaml change (partition_logs_by_resource_attributes → partition_logs_by_trace_id) is operationally significant but orthogonal to the Queue rewrite. Worth splitting into a separate commit/PR so it can be deployed or reverted independently.

🟢 appendRecord in test helper is O(n)

b {logs = Map.insert topic (existing <> [cr]) b.logs}

Appending to a list per record is quadratic over the seed. Fine for the 5-element seeds in spec, but worth a note if the test grows.

Minor nits (no action required)

• assigned (KE.assignment) is called twice per pollOnce when near the high/low watermarks. It's a local metadata read so it's cheap, but a single ps <- assigned at the top of pollOnce would be marginally cleaner.
• WorkItem as a 5-tuple is fine for an internal type; a named record would make pattern matches more self-documenting if this grows.

Overall the PR is in good shape. The correctness-critical invariants (watermark, flush-gate, never-past-failed) are well-tested, and the effect abstraction makes the test interpreter convincing. The main asks are restoring the dlqHeaders doctest executability and adding back the DLQ failure log context.

Code Review: PR #437 — kafka: unified Eff-based consumer + producer, tiered DLQ, flush-gated commits

The architectural direction is excellent: effects-based abstraction, pure logic separated from I/O, a correct flush-before-commit gate, and a well-structured tiered retry system. The PR description is thorough and the doctests on pure helpers are appreciated. Below are findings in rough priority order.

Correctness / Bugs

[High] IntSet stores Int, but Kafka offsets are Int64

data PartProgress = PartProgress {base :: !Int64, ahead :: !IntSet.IntSet}

Data.IntSet.IntSet stores Int. On 64-bit GHC Int happens to be 64 bits, but the type contract is wrong — fromIntegral :: Int64 -> Int silently narrows on any 32-bit target and is semantically unsound. The offset watermark is the data-loss-critical invariant. Use Data.Set.Set Int64 (or Data.IntSet.IntSet from a word64-containers variant) so the type reflects the domain.

[High] Non-exhaustive KafkaProducer match → runtime panic on library upgrade

_ -> error "runSharedKafkaProducer: unsupported KafkaProducer operation"

A new constructor in a future kafka-effectful bump causes a runtime panic in production rather than a compile-time error. Since the library is pinned by git hash, this is manageable short-term, but the pin comment in cabal.project should call this out explicitly, and the -Wincomplete-patterns flag should be enforced so GHC warns on upgrade.

[Medium] TOCTOU race on inflightVar / pausedVar reads in poll loop

The poll loop reads these two TVars via two separate readTVarIO calls, not inside one atomically block. Another thread could modify one between the reads. Wrap in:

(inflight, paused) <- atomically $ (,) <$> readTVar inflightVar <*> readTVar pausedVar

[Medium] committed / tracker maps grow unboundedly across rebalances

committedVar and trackerVar are never pruned. Partitions that leave the assignment on a rebalance stay in the maps forever. Prune stale entries using KE.assignment on each committer tick.

[Medium] Map.insertWith (const id) preserves stale state across rebalance

modifyTVar' trackerVar (Map.insertWith (const id) tpKey (PartProgress (minimum offs) mempty))

On re-assignment of a partition after rebalance, the old PartProgress is kept, potentially pinning the base offset lower than the actual committed offset. At minimum add a comment; ideally reset on reassignment by tracking the active assignment set.

[Low] minimum offs inside subChunksFor is partial by type

chunksByBytes guarantees non-empty chunks, so this never throws in practice, but the type doesn't enforce it. Returning NonEmpty from chunksByBytes would make minimum (from Data.List.NonEmpty) total and self-documenting.

Simplification / Code Quality

[Medium] WorkItem is a 5-tuple — use a named record

type WorkItem =
  ( (Text, K.PartitionId)
  , K.ConsumerRecord (Maybe ByteString) (Maybe ByteString)
  , [(Text, ByteString)]
  , [Int64]
  , Int
  )

A named record with RecordWildCards would make construction and destructuring sites unambiguous and significantly more readable. Counting tuple positions is error-prone.

[Medium] KE.assignment called twice per poll near watermark boundaries

when (inflight >= highWaterBytes && not paused) do
  ps <- assigned   -- first call
  ...
when (inflight <= lowWaterBytes && paused) do
  ps <- assigned   -- second call
  ...

Hoist ps <- assigned before the when guards (or use a single atomically read pattern) to avoid a redundant effect call in the common near-watermark case.

[Low] Duplicated BC.pack . toString pattern

bimap (BC.pack . toString) (BC.pack . toString)

This appears in at least two places (publishRawToKafka, consumerRecordHeadersToHashMap). Extract a shared:

textToBS :: Text -> BC.ByteString
textToBS = BC.pack . toString

and use bimap textToBS textToBS.

[Low] Magic number 64 for TBQueue bound

workQ <- newTBQueueIO 64

At ~64 MiB per chunk this is ~4 GiB of in-flight work. Add a named constant with a comment explaining the sizing.

[Low] sortOn (.crOffset) in subChunksFor may be redundant

pollMessageBatch from hw-kafka-client returns records in offset order per partition. If that guarantee holds, the sortOn is O(k log k) work done for free by the poll. Worth checking; removing it saves allocations on every poll.

[Low] b > 0 guard in committableCommits is confusing

The intent is "skip uninitialized partitions," but offset 0 is a valid commit target. The guard is correct for the Kafka semantics used here (base=0 means nothing completed yet), but a comment explaining why b > 0 rather than e.g. b > findWithDefault (-1) tp committed would prevent future misreads.

Test Coverage

[Medium] KafkaDlqReplay role path is not exercised

KafkaConsumerSpec only exercises KafkaPrimary. The DLQ replay path (including the monoscope-next-due-at header gating and the 5s min-delay backoff) is untested. A second integration test variant passing KafkaDlqReplay as the role would catch regressions in the retry tier routing.

[Medium] Retry tier escalation not tested

The unit doctests cover retryDestination for individual cases, but there is no test that a second-failure message actually lands on -retry-60s (vs the first-failure -retry topic) end-to-end through the integration harness.

[Low] QueueSpec.hs / doctest wiring

The PR description mentions QueueSpec but no such file appears in the diff. Confirm the doctests embedded in Queue.hs are wired into the CI test suite (e.g. via a doctests executable); otherwise they're documentation, not tests.

[Low] Integration test hard-codes DLQ topic name

let dlq = Map.findWithDefault [] "otlp_deadletter" b.logs

Derive from config instead:

let dlqTopic = tr.trATCtx.config.kafkaDeadLetterTopic

[Low] 15-second timeout backstop gives no diagnostic on expiry

drive returns silently after 15s if the done condition is never met; subsequent shouldBe assertions reveal the failure but with no timing context. A putStrLn "WARNING: drive timed out" or expectationFailure on timeout would cut debug time.

Security

[Low] Internal exception text in Kafka headers / HTTP responses

errorReason (via toText (show e)) is stamped into DLQ record headers verbatim and the same stringified KafkaError surfaces in HTTP responses from publishReplayEvent. Kafka headers are plaintext readable by any consumer; HTTP errors are visible to callers. Stack traces, SQL fragments, or internal hostnames could leak. The old code had the same pattern — but since this PR extends the exception-to-header path it's worth addressing: map to a structured error code + redacted message at the boundary.

Summary table

Overall this is a well-thought-out rewrite. The flush-before-commit gate is correct and the test that validates it is a nice piece of work. The high-priority items are type safety and the runtime-panic risk from the non-exhaustive match — both are small fixes. Recommend addressing those before merge.

Code Review — PR #437 (follow-up, post-12920404)

Several items from the prior rounds have already landed in the diff (WorkItem named record, Set Int64 in PartProgress, bimap in publishRawToKafka, unless (null commits && inflight == 0), atomic (inflight, paused) read, committer pruning on rebalance). Below is what still stands out in the current diff.

Unresolved from earlier rounds

takeWhile isDue ordering assumption — still no code comment (subChunksFor, KafkaDlqReplay branch)

The correctness of takeWhile (isDue nowEpoch) (sortOn (.crOffset) …) depends on monoscope-next-due-at being monotonically non-decreasing with offset within a partition (offset ≈ failure-time ≈ due-time). A single-line comment at the takeWhile call site would prevent future readers from breaking this silently. First raised in the first review, not yet addressed.

(cs,,) tuple-section in committer — still opaque

(cs,,) . Map.size <$> readTVar trackerVar <*> readTVar inflightVar

TupleSections + operator sectioning in an atomically do block is dense. Three explicit let/<- bindings are no longer and much more readable. Also mentioned in the first review, not yet addressed.

Remaining code quality

-- \| typo in dlqHeaders where-clause comment

    -- \| Headers stamped on every DLQ publish.

The backslash is spurious. Should be -- | (Haddock) or plain --. Small but ships into generated docs.

minimum work.offsets is partial by type

chunksByBytes guarantees non-empty chunks, so work.offsets is never empty in practice, but minimum :: [a] -> a is partial. minimumDef 0 work.offsets (from safe) or returning NonEmpty from chunksByBytes would make the invariant structural. The current code is safe, but one refactor away from a silent runtime error.

assigned potentially evaluated twice per pollOnce call

when (inflight >= highWaterBytes && not paused) do
  ps <- assigned   -- KE.assignment effect call
  …
when (inflight <= lowWaterBytes && paused) do
  ps <- assigned   -- second call
  …

In the case where both conditions fire (unlikely but possible if watermarks cross), KE.assignment is called twice. Hoisting ps <- assigned before both when guards avoids the redundancy and makes the two blocks share the same snapshot.

Summary

Architecture and tests remain solid. The remaining asks are all small: add a one-line comment on the takeWhile isDue invariant, simplify the (cs,,) section, fix the \| typo, and optionally hoist assigned.

Code Review

Well-designed refactor. The decoupled-loop architecture is a significant improvement: separating polling, processing, and committing into distinct threads eliminates the old single-threaded serialization bottleneck, the effectful abstraction makes the integration test credible, and the flush-before-commit invariant is correctly implemented and tested. The pure functions are clearly specified via doctests. Feedback below is focused on type safety, correctness nits, and dead weight.

WorkItem.offsets :: [Int64] should be NonEmpty Int64

minimum work.offsets (in decoupledLoop's pollOnce) is partial. The invariant is real — chunksByBytes always produces non-empty chunks, and DLQ items are singletons — but nothing in the type enforces it. The cleanest fix is to change chunksByBytes to return [NonEmpty a]:

chunksByBytes :: Int -> (a -> Int) -> [a] -> [NonEmpty a]

This propagates the guarantee upward, removes the minimum import entirely, and makes WorkItem.offsets :: NonEmpty Int64 safe without NE.fromList.

Data.List.Extra (lookup, minimum) imports are redundant

Both are identical to what Relude already provides — lookup is Prelude.lookup, minimum is Prelude.minimum. If chunksByBytes returns NonEmpty (above), the Data.List.Extra import can be dropped entirely.

dlqHeaders doctests won't run from a where clause

Moving dlqHeaders into publishToDeadLetterQueue's where block removes it from doctest discovery. The existing doctests (attempt-count bumping, left-biased <>) test a non-obvious invariant. Either:

• Promote dlqHeaders back to module scope (unexported, -- not exported), or
• Move the doctests into QueueSpec as explicit HSpec examples.

Wildcard error in runSharedKafkaProducer — tighten the grep hint

The cabal.project comment is good, but the grep target (runSharedKafkaProducer) is buried in prose. Adding a machine-greppable tag in the source makes bumps safer:

-- KAFKA_EFFECTFUL_UPGRADE: add new KafkaProducer constructors here
_ -> error "runSharedKafkaProducer: unsupported KafkaProducer operation"

Then cabal.project can say grep -r KAFKA_EFFECTFUL_UPGRADE instead of a file path that can drift.

recc in WorkItem for KafkaPrimary is the partition head, not the chunk head

The field is documented as "header-bearing record" but for KafkaPrimary it's recc :| _ from the unsorted byTP map — not the first record of the sorted chunk. This is harmless because attrsFor on primary derives ce-type from the topic name (not record headers), but it will confuse anyone who expects recc to be meaningful per-chunk. A one-line note on WorkItem.recc would help:

-- For KafkaPrimary: partition-head (headers unused; ce-type derived from topic).
-- For KafkaDlqReplay: the record itself.

workQueueBound comment overstates the ceiling

"caps in-flight work at ~4 GiB"

The actual backpressure ceiling is highWaterBytes = 4 × kafkaChunkTargetBytes = 256 MiB. The 4 GiB figure is the theoretical queue capacity only if all 64 slots are filled with maximum-sized chunks simultaneously — which the high-water pause prevents. The comment should say "secondary hard limit; primary backpressure ceiling is highWaterBytes (256 MiB)".

KafkaDlqReplay due-prefix assumption is undocumented

takeWhile (isDue nowEpoch) relies on offset order = due-time order within a partition. This holds because each retry tier is a separate topic and partition, so every record in one partition shares the same base delay (monotonically increasing publish timestamps → monotonically increasing due times). A one-liner on isDue would save a future reader from having to reconstruct this:

-- Safe to takeWhile: within one (topic, partition) all records share the same
-- retry tier delay, so due times increase monotonically with offset.

KE.flushProducer can block the committer for up to delivery.timeout.ms

Under broker degradation, flushProducer blocks up to delivery.timeout.ms (librdkafka default: 300 s). This stalls commits and metrics for the full timeout. hw-kafka-client's KP.flushProducer accepts a Timeout argument — consider passing a bounded value (e.g., K.Timeout 30_000) so the committer degrades gracefully rather than hanging. Stalled commits are safe (redelivery), but a 5-minute gap in the metrics log hides the outage signal the comment relies on.

Minor test nit

appendRecord's existing <> [cr] and b.pending <> [(topic, val, hdrs)] are O(n) per append. Fine for 5-record tests; if the spec ever grows to hundreds of records, switch to Data.Sequence or DList.

Positive callouts

• The integration test for flush-before-commit is exactly right — drop the gate and it fails. That's a good regression guard.
• completeOffsets' self-pruning invariant (ahead set can't grow without bound) is clearly stated and tested.
• Factoring ATBackgroundEffects out of ATBackgroundCtx so the Kafka loop can extend it is a clean layering.
• Removing pooledForConcurrentlyN + bracket in favour of Ki scoped threads is a straight simplification.

Code Review

Well-structured PR overall — the effectful abstraction makes testing genuinely tractable, the flush-before-commit gate is clearly motivated, and the pure-function / doctest approach for the core logic is exactly right. A few issues below, roughly ordered by severity.

1. Dead parameters in decoupledLoop (-Wunused-matches)

appLogger and tp are accepted but never used in the function body — all logging goes through the Log effect and tracing through Tracing, both already in ConsumerEff's ATBackgroundEffects stack. These should be dropped from the signature (and the call sites in kafkaService). As-is, GHC will warn here.

-- before
decoupledLoop appLogger appCtx tp role batchSize clientId fn = …

-- after
decoupledLoop appCtx role batchSize clientId fn = …

2. dlqHeaders doctests are silently inactive

dlqHeaders was previously a top-level binding with live doctests. After being moved to a where clause inside publishToDeadLetterQueue, those -- >>> examples are now unreachable by doctest — the function is never exported and where-bound bindings aren't scanned. The doctests look active but don't run: this is a test regression.

The simplest fix is to make dlqHeaders an unexported top-level definition. Alternatively, promote the logic into publishToDeadLetterQueue directly and move the examples to the top-level haddock.

3. import Data.List.Extra (lookup, minimum) reimports the partial minimum

Relude's minimum is safe ([a] -> Maybe a). This explicit import shadows it with the partial Prelude.minimum from extra. It's only used at one site:

-- Queue.hs:589
modifyTVar' trackerVar (Map.insertWith (const id) work.tpKey (PartProgress (minimum work.offsets) mempty))

work.offsets is always non-empty by construction (from subChunksFor), and since records are sorted ascending before chunking, minimum equals head. The cleanest fix is to make offsets :: NonEmpty Int64 in WorkItem and use NE.minimum or NE.head; this removes the Data.List.Extra import entirely (the lookup it imports is identical to Relude's). Alternatively, just use headDef 0 work.offsets or fromMaybe 0 (minimumMay work.offsets) and drop the import line.

4. bimap f f → join bimap f

-- Queue.hs:262
map (bimap (BC.pack . toString) (BC.pack . toString)) $ HM.toList attributes

-- simpler
map (join bimap (BC.pack . toString)) $ HM.toList attributes

Same pattern appears in consumerRecordHeadersToHashMap.

5. committableCommits: implicit left-bias in Map.fromList advanced <> committed

The intent (new advances overwrite stale committed entries) is correct, but Map.unionWith const makes the left-bias explicit and matches the standard idiom for "prefer the first map":

Map.union (Map.fromList advanced) committed

Minor readability nit, not a correctness issue.

Minor test note

appendRecord in KafkaConsumerSpec appends via existing <> [cr] (O(n)). Fine for the 5-record test fixture, but worth a DList or reverse-prepend if that helper ever gets reused at scale. Not a blocker.

Summary: items 1–3 are the ones worth resolving before merge. The architecture itself (decoupled loop, flush gate, tiered DLQ, in-memory broker test) is solid.

Code Review — PR #437: Unified Eff-based Kafka consumer + producer

Overall: this is a well-architected rewrite. The effect-layered design, pure decision functions, flush-gated commits, and structured-concurrency shell are all solid. The in-memory broker test is elegant. A few focused notes below.

Correctness / Bugs

fromRight [] silently swallows KafkaError on the PubSub path (src/Pkg/Queue.hs, pubsubService binding):

>>= \result ->
  liftIO
    $ fromRight []
    <$> runBackground … (runErrorNoCallStack @K.KafkaError (runSharedKafkaProducer appCtx (routeBatchOutcome …)))

routeBatchOutcome logs inside before throwing, but fromRight [] drops the error after the fact with no outer log or metric. A KafkaError here silently converts to an empty ack list (no commit), which is safe but invisible — ops won't see it in the log at the pubsubService call site. At minimum log the discarded Left before fromRight. The PR description acknowledges the PubSub durability gap as a known follow-up, but the silent discard is a separate issue.

Performance

size x called up to 3× per element in chunksByBytes:

go acc !n (x : xs)
  | n + size x > target, not (null acc) = reverse acc : go [x] (size x) xs
  | otherwise                            = go (x : acc) (n + size x) xs

Both branches call size x 1–2 times. With a strict let !s = size x before the guards this becomes 1 call per element — a minor but free win given the function runs over every record in every poll batch.

Safety

Wildcard _ -> error in runSharedKafkaProducer:

The cabal.project NB comment and the inline comment both call this out. The risk is real: a kafka-effectful bump silently adds a constructor and the process panics in prod. One mitigation worth considering: add a GHC {-# WARNING … #-} pragma or a custom unsupportedOp helper that formats the constructor name via show, so the panic message is at least informative at the crash site. The current message "runSharedKafkaProducer: unsupported KafkaProducer operation" won't include which constructor triggered it.

Style / Conciseness

Opaque (cs,,) . Map.size tuple section in the committer:

(cs,,) . Map.size <$> readTVar trackerVar <*> readTVar inflightVar

TupleSections is available, but (cs,,) with two trailing commas is non-obvious to readers. A plain do-bind or liftA2 (cs,,) (Map.size <$> readTVar trackerVar) (readTVar inflightVar) is equivalent and parses faster. Alternatively:

tracked  <- Map.size <$> readTVar trackerVar
inflight <- readTVar inflightVar
pure (cs, tracked, inflight)

Map.insertWith (const id) is correct but opaque:

const id = \_ old -> old (keep existing). Map.insertWithKey, Map.alter, or flip const would each be more idiomatic. Most readable:

modifyTVar' trackerVar $ \m ->
  case Map.lookup work.tpKey m of
    Nothing -> Map.insert work.tpKey (PartProgress (minimum work.offsets) mempty) m
    Just _  -> m

or use Data.Map.Strict.insertWith (flip const) which is at least a known idiom.

dlqHeaders moved into where but its doctest uses -- \|:

The \| escape prevents Haddock from treating it as a section header, which is correct for a where binding. But doctests in local where bindings aren't picked up by doctest unless the module is scanned at the top level. Verify your doctest runner actually exercises these; if not, promote dlqHeaders to a module-level unexported helper (like chunksByBytes) so the doctests run.

Test Coverage

• subChunksFor has an inline doctest but no entry in QueueSpec. Since it encodes both the DLQ "due prefix" and the primary byte-chunking, a few explicit unit tests for the edge cases (all due, none due, partial due, empty partition map) would improve confidence.
• committableCommits doctest only checks the b > 0 guard. A test where committed already has an entry and the watermark hasn't moved (no re-commit) would be useful.
• The KafkaConsumerSpec flush-before-commit test is excellent — the "drop the gate and it fails" comment makes the invariant verifiable.

Minor

• subChunksFor KafkaDlqReplay assumes sortOn (.crOffset) produces "due-time order". This holds as long as the DLQ topic is written in arrival order (which it is for a single producer), but a brief comment making this invariant explicit would help future readers.
• appendRecord in the test uses existing <> [cr] — O(n) per append. For test data (seedTopic bVar topic [0..4]) this is fine; just noting it for if tests scale.