Code Review

This PR fixes a real production incident: a no-ack chunk (transient write/DLQ failure → routeBatchOutcome []) pins the commit base forever because librdkafka won't redeliver uncommitted offsets in-session. The self-heal is sound in design — record the minimum un-acked offset in reseekVar, have the poll thread seek backward and reset the tracker. Three issues to address before merge, plus two test improvements.

1. reseekVar is not pruned on rebalance — correctness bug

src/Pkg/Queue.hs ~line 557–560

The committer prunes trackerVar and committedVar down to the live partition set on every tick, but reseekVar is never filtered:

-- existing
modifyTVar' trackerVar  (Map.filterWithKey \k _ -> S.member k live)
modifyTVar' committedVar (Map.filterWithKey \k _ -> S.member k live)
-- missing:
modifyTVar' reseekVar   (Map.filterWithKey \k _ -> S.member k live)

Without this, a partition that stalls (no-ack → reseekVar entry recorded) and is then revoked by a rebalance before the next pollOnce will cause seekPartitions to fire on a partition this consumer no longer owns. After the seek, line 594 unconditionally re-inserts the partition into trackerVar (Map.insert k (PartProgress o mempty)). Now the committer sees a tracker entry for a revoked partition, but asg never lists it again, so null asg is never true for it — the unless (null asg) guard skips the prune — and the entry grows stale forever, producing phantom metrics and potentially racing against the new owner's commits.

Fix: add the reseekVar filter line alongside the other two in the committer's unless (null asg) block.

2. minimum w.offsets → head w.offsets

src/Pkg/Queue.hs line 545

then modifyTVar' reseekVar (Map.insertWith min w.tpKey (minimum w.offsets))

w.offsets is built by fmap (.crOffset) over a chunk that was produced by chunksByBytes applied to records already sorted with sortOn (.crOffset). The list is always ascending, so minimum == head — and head is O(1) vs O(n). Data.List.Extra.minimum (the one pulled in by import Data.List.Extra (lookup, minimum)) is still the partial base minimum; both throw on an empty list, but head makes the sorted-prefix assumption explicit and avoids the full traversal. The identical situation exists at line 621 (minimum work.offsets).

-- before
(Map.insertWith min w.tpKey (minimum w.offsets))
-- after
(Map.insertWith min w.tpKey (head w.offsets))

3. Tracker reset — avoid intermediate list via Map.foldlWithKey' or (<>)

src/Pkg/Queue.hs line 594

atomically $ modifyTVar' trackerVar \tk ->
  foldl' (\m (k, o) -> Map.insert k (PartProgress o mempty) m) tk (Map.toList reseeks)

Map.toList reseeks allocates an intermediate [(k,o)] list just to fold it back into a Map. Two cleaner alternatives (pick either):

-- option A — no intermediate list, idiomatic Map combinator:
foldl (\m k o -> Map.insert k (PartProgress o mempty) m) tk reseeks
-- i.e. Map.foldlWithKey'

-- option B — highest altitude, reads as "reseek entries override tracker":
Map.map (`PartProgress` mempty) reseeks <> tk
-- (<>) on Map is left-biased union, so reseeks keys win — exactly what we want

Option B is the most concise and expresses the intent directly.

4. Test: hollow regression guard — missing attemptVar assertion

test/integration/Pkg/KafkaConsumerSpec.hs line 172

The test's only assertion is b.committed == Just 5, which is identical to the happy-path test. If the no-ack guard were accidentally removed (making the first call ack immediately), committed would still reach Just 5 and the test would pass. Add one line to prove the retry actually fired:

readTVarIO attemptVar >>= (`shouldSatisfy` (>= 2))

This simultaneously proves the reseek path was exercised (≥ 2 calls) and that it converged (the drive timeout wouldn't have expired with committed == Just 5).

5. Test mock: seekCursor drops partition ID — fidelity gap for multi-partition seeks

test/integration/Pkg/KafkaConsumerSpec.hs line 86

seekCursor c (K.TopicPartition (K.TopicName t) _ (K.PartitionOffset o)) = Map.insert t (fromIntegral o) c

cursor :: Map Text Int is keyed by topic only, so when production calls seekPartitions with two entries for the same topic but different partition IDs (both in reseekVar :: Map (Text, K.PartitionId) Int64), the mock applies only the last one — the other seek is silently dropped. Not triggered by any current test (all tests use one partition per topic), but it means any future multi-partition regression test would give false confidence. The mock's cursor and recordTp/committed use (topic, partitionInt) keys in committed but topic-only keys in cursor; making cursor :: Map (Text, Int) Int and threading the partition through seekCursor would close the gap without touching any existing test assertions.

Summary

Code Review

The fix correctly identifies the root cause (poll cursor advancing past un-acked offsets that librdkafka won't redeliver) and the reseek-on-no-ack approach is sound. A few issues worth addressing:

Bug — Non-atomic seek + tracker reset (Queue.hs:597-598)

KE.seekPartitions [...] (K.Timeout 5000)   -- IO, up to 5s
atomically $ modifyTVar' trackerVar \tk -> Map.map (`PartProgress` mempty) reseeks <> tk

There's a real race window between these two non-atomic steps. While seekPartitions is in-flight (up to 5 seconds), concurrent workers can complete acking batches for a reseek'd partition and advance the tracker. The subsequent left-biased <> merge overwrites their progress with PartProgress reseekOffset mempty, forcing those already-completed offsets to be reprocessed.

The result is duplicate processing rather than data loss (idempotent downstream), but under DLQ broker degradation this window is longest and most likely to trigger.

Notably, the test mock makes SeekPartitions instantaneous, so this race can't manifest in the regression test.

Simplest fix: reset the tracker before the seek (the seek failing is benign — the next poll cycle will retry), or accept the race explicitly in a comment and note it's covered by downstream idempotence.

Type safety — minimum w.offsets is partial (Queue.hs:545)

then modifyTVar' reseekVar (Map.insertWith min w.tpKey (minimum w.offsets))

offsets :: ![Int64] carries no type-level non-emptiness guarantee. Data.List.Extra.minimum [] throws ErrorCall — and this is outside the tryAny wrapping fn, so the exception propagates uncaught and kills the worker thread.

Currently safe by convention (chunksByBytes never emits empty chunks), but the invariant isn't enforced in the type. The same call existed at line 625 pre-PR; that's pre-existing debt, but line 545 is new and lands in a no-ack error path (higher-risk code path than the normal poll path).

-- safer — offsets are sorted ascending by construction, head is the minimum
then modifyTVar' reseekVar (Map.insertWith min w.tpKey (head w.offsets))
-- or promote offsets to NonEmpty Int64 to eliminate both call sites

Test — cursor type uses Int where Int64 is meant (KafkaConsumerSpec.hs:91)

seekCursor does fromIntegral o converting o :: Int64 to Int (the cursor map value type), while recordTp stores raw Int64 into committed. The narrowing is silent on 64-bit, but if a future test seeds large synthetic offsets (> maxBound :: Int32 on 32-bit CI), the cursor silently points to the wrong log index while the commit assertion still passes.

Changing cursor :: Map (Text, Int) Int → Map (Text, Int) Int64 (matching the committed map's value type) and dropping the fromIntegral o in seekCursor removes the inconsistency.

Nit — fromIntegral p is a no-op in seekCursor (KafkaConsumerSpec.hs:91)

seekCursor c (K.TopicPartition (K.TopicName t) (K.PartitionId p) ...) =
  Map.insert (t, fromIntegral p) ...  -- p :: Int already; fromIntegral is Int→Int

recordTp (line 89) uses (t, p) directly for the same (Text, Int) key type. The fromIntegral is a no-op and should be dropped for consistency.

Style — extra blank lines violate fourmolu.yaml (KafkaConsumerSpec.hs)

fourmolu.yaml sets newlines-between-decls: 2 (= one blank line between top-level declarations). Several hunks add two consecutive blank lines between declarations. fourmolu --mode check will flag these and produce noisy follow-up commits.

Code Review

The PR correctly diagnoses and fixes the 2026-06-25 commit-starvation wedge: on no-ack, record the lowest unacked offset and have the poll thread reseek, so librdkafka redelivers the un-committed chunk. The approach is sound and the regression test is well-structured. Three issues need attention before merge.

Bug 1 — seekPartitions exception permanently re-wedges the consumer (Critical)

src/Pkg/Queue.hs lines 595–597

reseeks <- atomically $ swapTVar reseekVar Map.empty   -- reseekVar is now empty
...
KE.seekPartitions [...] (K.Timeout 5000)               -- can throw

reseekVar is cleared before seekPartitions runs. If seekPartitions throws (5-second timeout, partition temporarily unassigned during rebalance, broker error), the reseek offsets are gone — reseeks is a local binding that is discarded and reseekVar stays empty. The partition's tracker base stays pinned and the consumer re-wedges silently, reproducing exactly the original bug.

Fix: wrap the seek in tryAny and re-merge failures back into reseekVar:

reseeks <- atomically $ swapTVar reseekVar Map.empty
unless (Map.null reseeks) do
  result <- tryAny $ KE.seekPartitions [...] (K.Timeout 5000)
  case result of
    Left _ -> atomically $ modifyTVar' reseekVar (reseeks <>)  -- restore, retry next poll
    Right _ -> do
      atomically $ modifyTVar' trackerVar \tk -> Map.map (`PartProgress` mempty) reseeks <> tk
      LogBase.logInfo "kafka.consumer.reseek" ...

Bug 2 — Tracker reset clears ahead, causing wider-than-intended redelivery (Medium)

src/Pkg/Queue.hs line 598

atomically $ modifyTVar' trackerVar \tk ->
  Map.map (`PartProgress` mempty) reseeks <> tk  -- wipes ahead for the partition

When concurrent workers process non-overlapping chunks of the same partition, a no-ack on chunk A (offsets 100–104) while chunk B (105–107) has already acked records into ahead={105,106,107} causes the tracker reset to clear those successfully-acked offsets. After redelivery and successful processing of 100–104, completeOffsets advances base to 105 then stops — ahead is empty, so 105–107 are never committed and get redelivered again.

Fix: when resetting the tracker for a partition, preserve ahead offsets that are strictly above the reseek base rather than discarding them:

Map.mapWithKey (\k o -> case Map.lookup k tk of
  Just (PartProgress _ fwd) -> PartProgress o (S.filter (> o) fwd)
  Nothing                   -> PartProgress o mempty) reseeks <> tk

Bug 3 — Stale reseek can seek a revoked partition (Medium)

src/Pkg/Queue.hs lines 557–564 vs. 595

The committer prunes reseekVar inside unless (null asg) (skipped when assignment is momentarily empty mid-rebalance). pollOnce drains reseekVar unconditionally via swapTVar. A reseek entry written just before a rebalance can survive the prune guard and trigger seekPartitions on a partition this consumer no longer owns, producing a Kafka error that crashes decoupledLoop.

Fix: filter reseeks against current assignment before calling seekPartitions, mirroring the same live-partition guard used for tracker/committed pruning:

asg <- assigned
let live = S.fromList [(t.unTopicName, p) | (t, p) <- asg]
    safeReseeks = Map.filterWithKey (\k _ -> S.member k live) reseeks
unless (Map.null safeReseeks) do
  KE.seekPartitions [... | ((t,p),o) <- Map.toList safeReseeks] (K.Timeout 5000)
  ...
-- put back anything filtered out so it retries after rebalance completes
unless (Map.null (reseeks `Map.difference` safeReseeks)) $
  atomically $ modifyTVar' reseekVar (reseeks `Map.difference` safeReseeks <>)

Simplification — seekCursor/recordTp two-equation wildcard helpers

test/integration/Pkg/KafkaConsumerSpec.hs lines 89–92

Both helpers use a two-equation pattern matching only PartitionOffset and silently swallowing every other constructor:

recordTp m (K.TopicPartition (K.TopicName t) (K.PartitionId p) (K.PartitionOffset o)) = Map.insertWith max (t, p) o m
recordTp m _ = m
seekCursor c (K.TopicPartition (K.TopicName t) (K.PartitionId p) (K.PartitionOffset o)) = Map.insert (t, fromIntegral p) (fromIntegral o) c
seekCursor c _ = c

A list comprehension collapses both into one-liners with no wildcard:

CommitPartitionsOffsets _ tps -> atomically $ modifyTVar' bVar \b ->
  b{committed = Map.unionWith max b.committed
      (Map.fromList [(t, p) o | K.TopicPartition (K.TopicName t) (K.PartitionId p) (K.PartitionOffset o) <- tps])}
SeekPartitions tps _ -> atomically $ modifyTVar' bVar \b ->
  b{cursor = Map.union
      (Map.fromList [((t, fromIntegral p), fromIntegral o) | K.TopicPartition (K.TopicName t) (K.PartitionId p) (K.PartitionOffset o) <- tps])
      b.cursor}

This removes both local helpers, matches how the upstream type changes propagate automatically to a type error instead of a silent wildcard match, and aligns with the project's preference for conciseness.

Minor — <> vs Map.union on line 598

Map.map (\PartProgress` mempty) reseeks <> tkrelies on the comment to explain left-biased merge.Map.union (Map.map ...) tk` makes the intent self-documenting without the comment.

Code Review

The fix is conceptually sound: on no-ack, record the minimum un-acked offset per partition in a shared TVar, and have the poll thread seek back and reset the tracker so the chunk is redelivered. The regression test correctly guards against the fix being silently removed. A few issues to address:

Bug: reseek entries permanently lost on seekPartitions failure

src/Pkg/Queue.hs ~line 574–577

reseeks <- atomically $ swapTVar reseekVar Map.empty   -- ① clears atomically
unless (Map.null reseeks) do
  KE.seekPartitions [...] (K.Timeout 5000)              -- ② IO, can throw
  atomically $ modifyTVar' trackerVar ...               -- ③ separate STM tx

swapTVar atomically drains reseekVar at ①. If seekPartitions at ② throws (broker timeout, KafkaError, network blip), execution unwinds before ③ ever runs. reseekVar is empty and trackerVar is not reset — the self-heal is silently skipped. No future poll iteration will reseek because nothing writes to reseekVar again (the original no-ack chunk has already been processed). The partition base stays pinned exactly as before the fix, requiring an operator restart to unstick. The fix defeats itself on the class of transient errors it's meant to handle.

Fix: restore reseekVar on failure before re-throwing:

reseeks <- atomically $ swapTVar reseekVar Map.empty
unless (Map.null reseeks) do
  seekResult <- tryAny $ KE.seekPartitions [...] (K.Timeout 5000)
  case seekResult of
    Left err -> do
      atomically $ modifyTVar' reseekVar (reseeks <>)  -- put them back
      LogBase.logAttention "kafka.consumer.reseek.failed" $ AE.object [...]
    Right () -> do
      atomically $ modifyTVar' trackerVar \tk -> Map.map (`PartProgress` mempty) reseeks <> tk
      LogBase.logInfo "kafka.consumer.reseek" $ AE.object [...]

Bug: minimum w.offsets is partial — crashes on empty list

src/Pkg/Queue.hs ~line 543

then modifyTVar' reseekVar (Map.insertWith min w.tpKey (minimum w.offsets))

Prelude.minimum on [] throws inside the STM block. The same pattern exists at the pre-existing line 596, so this is consistent — but the new no-ack path is a new caller with the same latent crash. WorkItem.offsets is guaranteed non-empty by chunksByBytes today, but the type is [Int64] with no enforcement. Both call sites should use NE.minimum (after establishing w.offsets as NonEmpty Int64) or minimumDef from safe. If changing the type is out of scope for this PR, at minimum add a case nonEmpty w.offsets of Nothing -> pass; Just ne -> ... guard on the new branch.

Wrong cabal stanza

monoscope.cabal ~line 1165

 test-suite unit-tests
   ...
+    , base

The new test (KafkaConsumerSpec.hs) is in test/integration/, compiled under integration-tests, not unit-tests. If base was genuinely missing it should be added to the integration-tests stanza (or package.yaml equivalent). Adding it to unit-tests is a no-op for the new test and a spurious dep for unit-tests.

Redundant import

src/Pkg/Queue.hs ~line 21

import Control.Concurrent.STM (swapTVar)

unliftio (already depended on throughout this file for atomically, modifyTVar', etc.) re-exports swapTVar via UnliftIO.STM. This explicit Control.Concurrent.STM import is unnecessary — remove it and the call site just works.

Comment density

Four multi-line block comments were added inside decoupledLoop (above reseekVar, above the if null acks, above the revoked-partition filter, and above pollOnce's reseek block). Three of them restate the same causality chain (poll cursor moves past un-acked offsets, librdkafka won't redeliver, base pins). The PR description covers this fully. Inline, a one-liner pointing at reseekVar is sufficient:

-- no-ack: record lowest offset for poll thread to reseek (see reseekVar)

The trailing -- left-biased: reseek bases override on line 577 describes standard Map.<> behaviour — drop it.

Given the project's stated priority of keeping the codebase small, trimming ~15 comment lines here would be consistent with that goal.