Code Review — ingestion: TF-strict commit, per-partition offsets, row-level bisection

The durability semantics are sound and the design rationale is well-documented. A few code-quality and succinctness notes below.

Pkg/IngestError.hs — unnecessary indirection

Every call site writes IE.isPoison (IE.classify e). The two exported predicates (isTransient, isPoison) add a layer without benefit; isTransient isn't even used in this PR. Collapse them into direct exception predicates:

-- replaces isTransient . classify / isPoison . classify at call sites
isTransientException, isPoisonException :: SomeException -> Bool
isTransientException e = maybe False Hasql.isTransientHasqlError (fromException e)
isPoisonException = not . isTransientException

This removes the IngestError sum type entirely, mirrors the existing maybe False Hasql.isTransientHasqlError (fromException e) pattern already used in bulkInsertOtelLogsAndSpans:go, and makes call sites one function call instead of two.

Pkg/Queue.hs — trivial alias

allTopics = kafkaTopics   -- line 217

This is now a single-element alias for kafkaTopics with no semantic content (previously it included the DLQ topic). Inline it into consumerSub and remove the binding.

publishToDeadLetterQueue — verbose result assembly

pure $ case lefts results of
  [] -> Right ()
  (err : _) -> Left err

This is maybe (Right ()) Left . listToMaybe $ lefts results, which is a one-liner:

pure . maybe (Right ()) Left . listToMaybe $ lefts results

bulkInsertOtelLogsAndSpans — bisection depth cap is over-provisioned

The cap is 12 levels (2^12 = 4096 rows), but rawInsert already chunks at 700 rows, so the maximum vector reaching go is 700. A cap of 10 covers 1024 — more than enough. Minor, but aligns the constant with reality.

Also: when depth is exhausted with >1 rows (d <= 0 branch), the error is re-thrown, stalling the partition. Given chunkSize = 700 < 2^10, this branch should be unreachable in practice, but it silently contradicts the "a bad row no longer wedges the partition" goal in the PR description. Worth a comment (or a logAttention "BISECT_DEPTH_EXHAUSTED" before the re-throw) so on-call knows it's theoretically possible.

bulkInsertOtelLogsAndSpansTF — guard ordering

(_, Left eTf) -> throwIO eTf

The wildcard catches (Right _, Left eTf) (TF failed, PG succeeded — correct, abort) but also (Left _, Left eTf) (both failed). In the double-failure case the PG error is silently swallowed. Consider logging ePg here too — it's useful triage context when a TF outage is compounded by a PG outage.

Minor

• IngestError.hs imports Relude but uses nothing from it beyond what SomeException/fromException need — those come from base. The import is harmless but noise if Relude re-exports base's exception hierarchy.
• bulkInsertOtelLogsAndSpansTF: the haddock references CLAUDE.md "Storage migration" — that note lives outside the repo and will rot. A brief inline note (-- PG is temporary; see PR #402 or migration runbook) ages better.

Overall the design is solid. The main wins would be collapsing IngestError into direct SomeException -> Bool predicates (removes a module and simplifies call sites) and inlining allTopics.

Code Review

The core logic is sound and the durability semantics are well-reasoned. A few concrete improvements below.

Bugs / Correctness

None found — the per-partition commit, bisect, and TF-strict semantics all look correct.

Repeated pattern — extract a helper (Data.Effectful.Hasql)

maybe False Hasql.isTransientHasqlError (fromException e) appears 4 times across two files. Data.Effectful.Hasql already exports isTransientHasqlError; add one line there and all call sites collapse:

-- in Data.Effectful.Hasql (exported)
isTransientException :: SomeException -> Bool
isTransientException = maybe False isTransientHasqlError . fromException

Call sites become Hasql.isTransientException e — no maybe False … fromException boilerplate anywhere.

publishToDeadLetterQueue — prefer partitionEithers (already in scope)

partitionEithers is already used at line 235 of this same file. The final line:

pure . maybe (Right ()) Left . listToMaybe $ lefts results

reads more clearly as:

let (errs, _) = partitionEithers results
pure $ maybe (Right ()) Left (listToMaybe errs)

Both are equivalent; this just keeps the idiom consistent with the rest of Queue.hs.

IORef accumulator for commitTPs — can be a plain list

The commitTPs IORef exists only because recordSuccess is called inside a forM_ body. forM_ discards the element results, but you could flip it to fmap concat . forM and have recordSuccess return the TPs instead of writing to an IORef:

tps <- fmap concat $ forM (HM.toList byTopic) \(topic, neRecords@(recc :| _)) -> do
  ...
  case result of
    Right _ -> pure (tpsFor neRecords topic)
    Left e -> do
      ...
      if isTransient
        then pure []
        else do
          dlqRes <- ...
          pure $ case dlqRes of { Right _ -> tpsFor neRecords topic; Left _ -> [] }

where tpsFor extracts the two-liner currently inside recordSuccess. Removes the IORef, newIORef, modifyIORef', readIORef, and liftIO wrapping.

Bisect depth: PR description inconsistency

PR description says "Recursion capped at 12 levels"; code is go (10 :: Int) and the Haddock says "10 levels (1024 rows)". The code and docstring agree; the PR description is wrong. Fix the description or the constant (12 levels = go (12 :: Int) = 4096 leaf cap, safely above chunkSize = 700).

IngestError.hs mentioned but not in diff

PR description references a new Pkg/IngestError.hs (IngestError = TransientInfra | PoisonBatch + classify) that replaces string-matching. The file is not in this diff and doesn't exist in the repo. Either it was dropped or the description is stale — worth clarifying so reviewers aren't confused about whether classification is being deferred or already done via isTransientHasqlError.

Minor

• modifyIORef' commitTPs (tps ++) prepends each batch's TPs (result list is reversed relative to processing order). No correctness impact since commitPartitionsOffsets doesn't care about order, but worth noting.
• The threadDelay removal is documented in the PR description only; a one-line commit message or inline comment on the poll call would prevent future readers from re-adding it as a "fix".

Code Review

Solid durability work — the three-bug root cause is well diagnosed and the fix is coherent. A few conciseness/idiom notes given the project's priority on keeping the codebase small:

fmap concat $ forM → concatMapM

extra is already a dependency and exports Control.Monad.Extra.concatMapM. Queue.hs:255 can be:

-- before
tps <- fmap concat $ forM (HM.toList byTopic) \(topic, neRecords@(recc :| _)) -> do ...

-- after
tps <- concatMapM (\(topic, neRecords@(recc :| _)) -> ...) (HM.toList byTopic)

partitionEithers + ignored _ → lefts

publishToDeadLetterQueue only needs the error side. Data.Either.lefts (re-exported via Relude) is more direct and avoids the discarded binding:

-- before
let (errs, _) = partitionEithers results
pure $ maybe (Right ()) Left (listToMaybe errs)

-- after
pure $ maybe (Right ()) Left $ listToMaybe $ lefts results

0 <$ action formatting

Telemetry.hs (POISON_ROW_DROPPED branch) puts 0 alone on its own line before <$. The conventional operator for this direction is $>:

-- before
0
  <$ Log.logAttention "POISON_ROW_DROPPED" ...

-- after
Log.logAttention "POISON_ROW_DROPPED" ... $> 0

Reads left-to-right, same semantics.

isTransientException doc comment

Per project style: default to no comments; only add one when the WHY is non-obvious. The body maybe False isTransientHasqlError . fromException is self-explanatory. The comment currently explains WHAT, not WHY it exists as a separate export. Either drop it or trim to a single line:

-- Exported so call sites avoid repeating @maybe False isTransientHasqlError . fromException@.
isTransientException :: SomeException -> Bool

Logic / correctness

• Bisection depth cap: 10 levels handles 2^10 = 1024 rows, which safely covers chunkSize = 700. Correct.
• Guard ordering in go: V.length recs == 1 fires before d <= 0, so a single poison row is always DLQ'd even at depth 0. Intended, and correct.
• TF-thread result flow: the fork catches via tryAny, logs internally, then pure res surfaces the Either to the outer case (pgRes, tfRes) — no double-log, no scope teardown race. Correct.
• Per-partition commit gate: unless (null tps) correctly skips the commit call when all batches failed, avoiding an empty commit that would silently advance nothing but still hit librdkafka. Correct.
• DLQ unsubscribed: removing allTopics and using only kafkaTopics is the right call; the comment explaining why is one of the genuinely non-obvious WHY cases that justifies keeping it.

Minor

• tpsFor is defined as a let-binding inside the do-block capturing topic/neRecords as explicit arguments. It reads clearly there; no need to move it to where, but it could go there if future callers emerge.
• The isUnrecoverableError removal is correct — string-matching on show e was fragile and now superseded by Hasql.isTransientException with proper exception hierarchy classification.

Overall the change is well-reasoned. The four items above are all line-count/idiom improvements rather than correctness issues.

Code Review

Good structural fix — per-partition commits and TF-strict semantics are the right direction after that incident. A few things need attention before merging.

Bug: bisection + non-idempotent INSERT = duplicate PG rows (non-TF path)

src/Models/Telemetry/Telemetry.hs, bulkInsertOtelLogsAndSpans

Bisection is only entered on a non-transient failure. But once inside, the <*> in:

(+) <$> go (d - 1) l <*> go (d - 1) r

is sequential. If go (d-1) l successfully inserts its rows, then go (d-1) r hits a transient error (connection reset, pool exhaustion), the transient exception propagates all the way out. In non-TF mode (else void $ bulkInsertOtelLogsAndSpans records), the caller's tryAny sees a transient error → isTransientException → pure [] → no commit → Kafka redelivers the full batch. On redelivery mintOtelLogIds generates fresh UUIDs and the full batch is inserted again, including the already-written left half. INSERT INTO otel_logs_and_spans has no ON CONFLICT, so duplicates land.

In TF-strict mode this doesn't cause duplicates (TF success commits the offset), but the non-TF code path still exists and has this invariant broken.

The docstring says "Transient Hasql errors propagate so the surrounding retry logic can act" — but that surrounding retry re-sends the whole batch, not just the right half.

Bug: partial DLQ failure stalls already-DLQ'd partitions and causes duplicate DLQ entries

src/Pkg/Queue.hs, publishToDeadLetterQueue

pure $ maybe (Right ()) Left $ listToMaybe $ lefts results

If a batch of N messages DLQ-publishes N-1 successfully and the last one fails, this returns Left. Both callers map Left _ -> [], so no offsets are committed. On the next poll, all N messages are redelivered — the N-1 already-DLQ'd messages are sent to DLQ again. This loops until the DLQ publish fully succeeds, making duplicate DLQ entries guaranteed on any partial failure. At minimum this should be documented; at worst it inflates poison traffic and confuses replay tooling.

otelRowSnippet called O(log N) times per row during bisection

src/Models/Telemetry/Telemetry.hs, rawInsert

rawInsert calls V.mapM otelRowSnippet recs on every invocation. otelRowSnippet has observable effects (logs OTEL_UNPARSEABLE_IS_REMOTE, etc.). A batch that bisects K levels deep will call otelRowSnippet on each row up to K times — inflating those log counts by up to 10× and making the alert signal noisy. Pre-computing snippets outside go and threading them through would eliminate this; alternatively a note in the docstring that snippet effects are repeated would at least make the tradeoff visible.

Conciseness: publishToDeadLetterQueue final expression

The current:

results <- forM messages \... -> do
  ...
  pure (Left err)  -- or pure (Right ())
pure $ maybe (Right ()) Left $ listToMaybe $ lefts results

can fold the results binding away entirely:

fmap sequence_ $ forM messages \... -> do
  ...
  pure (Left err)  -- or pure (Right ())

sequence_ :: [Either Text ()] -> Either Text () traverses with Either's Applicative and returns the first Left or Right (), which is exactly the current semantics. Saves a binding and the lefts/listToMaybe chain.

Minor: tpsFor offset arithmetic

Map.fromListWith max [(r.crPartition, K.unOffset r.crOffset + 1) | r <- toList neRecords]

K.unOffset on a sentinel offset like OffsetEnd (-1) gives -1; +1 gives 0, committing partition to the beginning on next startup. This is probably unreachable in practice (librdkafka delivers concrete offsets only), but worth a guard or assertion to make the assumption explicit.

Not a bug (confirming): concatMapM dependency

Control.Monad.Extra (concatMapM) — extra is already in build-depends. Fine.

Not a bug (confirming): isTransientException vs AnnotatedException

The isTransientException e check in both services operates on the exception caught by an inner tryAny that fires before any checkpoint annotation wrapper runs, so fromException @HasqlException sees the unwrapped type. No issue here.

Code Review

Solid fix for a real data-loss incident. The core logic is correct. A few observations below, focused on succinctness and correctness.

publishToDeadLetterQueue — fmap sequence_ $ forM

fmap sequence_ $ forM messages \(origTopicOrAckId, msgData) -> do

This is functionally correct (forM runs all IO effects before sequence_ folds the pure Either list), but the idiom is opaque. The explicit bind form is clearer and matches the style used elsewhere in the file:

results <- forM messages \(origTopicOrAckId, msgData) -> do
  ...
pure (sequence_ results)

Or even shorter with mapM:

results <- mapM (publishOne currentTime) messages
pure (sequence_ results)

Data.Map.Strict for tpsFor

Map.fromListWith max works, but if K.PartitionId has a Hashable instance (hw-kafka-client derives it), HM.fromListWith max keeps it consistent with the rest of Queue.hs and avoids the Data.Map.Strict import.

Unreachable guard in bulkInsertOtelLogsAndSpans

go _ pairs | V.null pairs = pure 0

This guard is unreachable in normal flow: the outer function guards V.null records before calling go, and the only recursive calls split a vector of length ≥ 2 (the V.length pairs == 1 case is handled before recursion). Consider removing it or adding a comment if it's intentionally defensive.

Sequential <*> in bisection

(+) <$> go (d - 1) l <*> go (d - 1) r

In Eff es, <*> is sequential. If go l throws (transient or depth-exhausted), go r is skipped. This is correct for transients (want whole batch to fail). For depth exhaustion it means the right half never attempts its rows — but that's documented and acceptable given the cap is well above chunkSize. No change needed, but a comment noting the sequential semantics would prevent a future refactor from accidentally introducing parallelism here.

Test coverage

The PR acknowledges tier-2 tests are deferred, but the five regression scenarios listed in the PR body (kafka_pg_down_tf_up_commits, poison_row_quarantines_not_stalls, dlq_publish_fail_blocks_commit, etc.) directly guard the behavior this PR changes. Worth at minimum opening a tracking ticket before this merges, so the test debt doesn't get lost.

Minor positives worth keeping

• isTransientException consolidation is a clean DRY win — the four-site duplication of maybe False isTransientHasqlError . fromException is gone.
• Removing isUnrecoverableError (string-match heuristic) is the right call; structured exception types are more reliable.
• tpsFor guard K.unOffset r.crOffset >= 0 correctly filters librdkafka sentinel offsets.
• $> 0 for POISON_ROW_DROPPED is idiomatic.
• Removing the DLQ topic from live consumer subscription is a good operational improvement.

Code Review

Summary

Fixes three intertwined durability bugs: offset commit regardless of batch outcome, DLQ self-subscription, and the brittle string-match error classifier. Core correctness logic is sound. Notes below are mostly succinctness / minor nits.

Data/Effectful/Hasql.hs

Good. isTransientException elimination of maybe False isTransientHasqlError . fromException at call sites is exactly the right abstraction.

One missed call site remains in Telemetry.hs:879:

let isHasql = isJust (fromException @Hasql.HasqlException e)

This is checking "any Hasql exception" (not just transient ones), so isTransientException doesn't apply — but you could export isHasqlException :: SomeException -> Bool from Hasql to give this site the same treatment. Minor, separate PR material.

Models/Telemetry/Telemetry.hs

bulkInsertOtelLogsAndSpansTF

The (pgRes, tfRes) case-split is clean and the TF-strict contract is clearly expressed. One observation: when both writes fail, TIMEFUSION_WRITE_FAILED/TIMEFUSION_WRITE_BUG fires from tfThread and BOTH_WRITES_FAILED fires from the outer combiner — two log entries for one event. That's acceptable for alerting (different log IDs), but monitoring dashboards should dedupe by trace/correlation ID to avoid double-counting.

bulkInsertOtelLogsAndSpans — bisection go

Guard ordering is correct: V.length pairs == 1 intentionally fires before d <= 0, dropping the poison row regardless of depth. The inline comment is the right call to guard against future reordering.

The docstring has grown to ~15 lines. The hoisting-rationale paragraph and the known-edge-case block are PR-description material, not per-call documentation. Suggested trim:

-- | Hasql bulk insert for OtelLogsAndSpans, chunked to fit the 65535-param PG
-- bind limit (700 * 88 = 61600). On a non-transient failure bisects to isolate
-- poison rows (cap: 10 levels). Transient errors propagate for the caller to retry.
-- Row snippets are built once above 'go'; otelRowSnippet effects fire exactly once.
bulkInsertOtelLogsAndSpans :: ...

rawInsert builds chunks via let ... in case. Moving the chunkStmt binding into a where clause of rawInsert avoids the awkward let ... in case nesting:

rawInsert pairs =
  case chunks of
    [single] -> Hasql.session (HSession.statement () (chunkStmt single))
    _ -> Hasql.transaction TxS.ReadCommitted TxS.Write $ Relude.sum <$> traverse (Tx.statement () . chunkStmt) chunks
  where
    chunkSize = 700
    chunkStmt c = dynamicallyParameterized (otelInsertHeader <> mconcat (intersperse ", " (V.toList c))) D.rowsAffected True
    chunks = unfoldr (\v -> if V.null v then Nothing else Just (V.splitAt chunkSize v)) (V.map snd pairs)

Pkg/Queue.hs

publishToDeadLetterQueue

Signature change to IO (Either Text ()) is the right fix — a broken DLQ was silently dropping poison data.

sequence_ on the already-collected [Either Text ()] is correct: all forM IO actions run eagerly, sequence_ just folds the results to the first Left. The partial-failure trade-off (duplicate DLQ entries vs. data loss) is documented and correct.

Succinct alternative (same semantics, no intermediate results binding):

publishToDeadLetterQueue appLogger appCtx messages attributes errorReason = do
  let deadLetterTopic = appCtx.config.kafkaDeadLetterTopic
  currentTime <- getCurrentTime
  fmap sequence_ $ forM messages \(origTopicOrAckId, msgData) -> do
    ...
    publishJSONToKafka appCtx deadLetterTopic (BC.unpack msgData) deadLetterAttrs >>= \case
      Left err -> runLogT "dlq" appLogger LogAttention
        (LogBase.logAttention "publishToDeadLetterQueue: failed to publish"
         $ AE.object ["error" AE..= err, ...])
        $> Left err
      Right _ -> pure (Right ())

kafkaService — tpsFor / processGroup

Map.fromListWith max for per-partition max-offset is the right tool. concatMapM processGroup from extra is correct and idiomatic.

One nit: tpsFor closes over topic as a parameter, but the final list comprehension duplicates K.TopicName topic per element. map (uncurry (K.TopicPartition (K.TopicName topic))) . Map.toList avoids repeating the constructor:

tpsFor topic neRecords =
  map (uncurry (K.TopicPartition (K.TopicName topic))) . Map.toList
    $ Map.fromListWith max
      [(r.crPartition, K.unOffset r.crOffset + 1) | r <- toList neRecords, K.unOffset r.crOffset >= 0]

Removal of isUnrecoverableError (string-matching on show e) is a meaningful correctness improvement — unknown exceptions now route to DLQ rather than stalling the partition silently.

Test coverage

The PR acknowledges tier 2 tests are deferred. The five regression scenarios (kafka_pg_down_tf_up_commits, kafka_tf_down_does_not_commit, poison_row_quarantines_not_stalls, dlq_publish_fail_blocks_commit, per_partition_commit_isolates_healthy_topics) are exactly right and should be the first thing in the follow-up PR before another production incident makes them feel urgent.

Overall

The correctness improvements (per-partition commits, TF-strict contract, bisection, isTransientException) are solid. Main ask before merging: trim the bulkInsertOtelLogsAndSpans docstring and open a tracking issue for the tier 2 test harness so it doesn't slip.

Code Review

Overview: Fixes the 2026-06-01 data-loss incident with three interlocking changes — per-partition Kafka commits, TF-strict dual-write semantics, and row-level bisection for poison isolation. The design rationale is well-documented and the logic is sound. A few improvements worth considering:

Succinctness / idiomatic style

tfThread body — use whenLeft_ (8+ existing call sites in this codebase):

-- current (5 lines)
case res of
  Left e -> do
    ...
  Right _ -> pass
pure res

-- simpler (3 lines)
whenLeft_ res \e -> do
  ...
pure res

publishToDeadLetterQueue — sequence_ results swallows errors 2..N

sequence_ on [Either Text ()] short-circuits at the first Left, so only the first DLQ-publish failure is returned; subsequent failures are only in the log. For the caller's purpose ("did any fail?") this is fine, but it's subtly different from the partitionEithers + listToMaybe the PR description claims. Consider a note in the code:

-- sequence_ short-circuits at the first Left; subsequent failures are logged above.
pure (sequence_ results)

Or rename to firstFailure = sequence_ results at the let binding level for clarity.

Bisection depth — implicit contract with batchSize

go 10 guarantees single-row isolation only when V.length pairs ≤ 2^10 = 1024. The default messagesPerPubsubPullBatch = 200 is safe, but if it were raised above 1024, BISECT_DEPTH_EXHAUSTED would fire before isolating the poison row and the whole batch would be re-queued indefinitely.

The docstring says "comfortably above chunkSize = 700" but the relevant bound is the batch size, not chunkSize. Suggest:

-- cap: 10 levels → isolates single rows for batches ≤ 1024.
-- messagesPerPubsubPullBatch defaults to 200; raise the cap if that grows past 1024.
go (10 :: Int) pairs

Or a cheap runtime guard: when (V.length pairs > 1024) $ Log.logAttention "BISECT_CAP_MAY_NOT_ISOLATE" ...

Correctness

rawInsert atomicity under bisection — correct. Each rawInsert call is atomic (single multi-row INSERT or a transaction for multiple chunks). A transient from go (d-1) r after go (d-1) l succeeds propagates upward, but since pgThread wraps the whole thing in tryAny and TF is the commit signal, the partial PG write is harmless under TF-strict semantics.

tpsFor offset arithmetic — correct. K.unOffset r.crOffset + 1 is standard Kafka committed-offset semantics (next-to-fetch), and >= 0 correctly filters OffsetEnd/OffsetInvalid sentinels.

Guard ordering in go — correct. V.length pairs == 1 precedes d <= 0, so a single-row poison at depth 0 is dropped as poison (correct) rather than re-raised as depth-exhausted.

(,) <$> Ki.await pgThread <*> Ki.await tfThread — correct. Both threads use tryAny internally so they never throw; Ki.atomically blocks until both resolve cleanly.

Minor

• Column count 700 * 88 = 61600 is accurate (I counted 88 entries in otelColumns). The old comment's "87 columns" was wrong — good fix.
• Removing the DLQ subscription from the live loop is the right call; the comment explaining the rationale is clear.
• Removal of isUnrecoverableError (string-matching) in favour of the typed Hasql.isTransientException is a clear improvement.

Code Review

This is a solid durability fix for a real production incident. The architecture decisions (TF-strict semantics, per-partition commits, bisection isolation) are well-reasoned and the PR description is exemplary. A few focused notes below.

Correctness / Bugs

publishToDeadLetterQueue — sequence_ semantics are correct but surprising

let firstFailure = sequence_ results
pure firstFailure

sequence_ on [Either Text ()] short-circuits on the first Left using the Either monad — which is correct here because forM already ran all IO effects and collected all results before sequence_ touches them. The comment explains it, but the idiom is still surprising to a reader who equates sequence_ with "run actions sequentially."

Simplify to one line and make the intent clearer:

pure $ sequence_ results
-- or, more explicit:
pure $ maybe (Right ()) Left $ find isLeft results >>= either Just (const Nothing)

Or even simpler — since sequence on Either IS the right tool, just drop the let:

pure (sequence_ results)

Code Succinctness

tpsFor — point-free with uncurry … . second … is harder to read than a list comprehension

-- current
tpsFor topic neRecords =
  map (uncurry (K.TopicPartition (K.TopicName topic)) . second K.PartitionOffset)
    $ Map.toList
    $ Map.fromListWith max [...]

A list comprehension is shorter and clearer:

tpsFor topic neRecords =
  [ K.TopicPartition (K.TopicName topic) p (K.PartitionOffset o)
  | (p, o) <- Map.toList $ Map.fromListWith max
      [(r.crPartition, K.unOffset r.crOffset + 1) | r <- toList neRecords, K.unOffset r.crOffset >= 0]
  ]

concatMapM import from Control.Monad.Extra — used exactly once

extra is already a dep, so the import is fine, but concat <$> mapM processGroup (HM.toList byTopic) or fmap concat . forM (HM.toList byTopic) $ processGroup avoids the new qualified import for a one-liner. Minor, but aligns with "fewer lines" priority.

Two adjacent let blocks in kafkaService — tpsFor and processGroup are in separate let blocks. They can be merged into one, or tpsFor can be moved to the where clause of kafkaService (it doesn't close over anything from the inner do).

Behavioral Change Worth Calling Out

isUnrecoverableError removal — the old function matched on error strings:

• "Unknown wire type" (protobuf)
• "project API Key and project ID not available" (auth)
• "Unexpected end of input" / "Invalid UTF-8 stream" (encoding)

These are now classified as else (non-transient Hasql → poison → DLQ) in both kafkaService and pubsubService. This is semantically more correct (they are poison rows, not infra failures), but the behavioral change means those errors now block on DLQ publish success before acking. If the DLQ publish fails for one of these (e.g. Kafka auth error AND DLQ Kafka auth error), the partition stalls instead of silently discarding. That's the right trade-off, but worth a brief note in the PR checklist or a log message distinguishing "non-Hasql poison" from "Hasql poison."

Minor Nits

• Relude.when (V.length records > 1024) $ Log.logAttention "BISECT_CAP_MAY_NOT_ISOLATE" … — the Relude. qualifier is only needed if when is ambiguous; Relude is imported unqualified so when alone should work.
• The -- <*> is sequential in Eff es comment in go is good defensive documentation. Consider also putting a note in rawInsert's name that it does not catch exceptions (so go can tryAny it correctly).
• The isTransientException export in Hasql.hs is a clean DRY win. The Haddock comment is accurate.

Summary

The core logic is correct and the durability improvements are significant. Main asks:

1. Simplify let firstFailure = sequence_ results; pure firstFailure → pure $ sequence_ results.
2. Replace uncurry … . second … in tpsFor with a list comprehension.
3. Document or log the behavioral difference for non-Hasql poison errors (previously silently discarded, now DLQ-gated).

Code Review

The durability model is sound and the design rationale is well-documented. Three succinctness notes and one operational concern.

Pkg/Queue.hs — concat <$> mapM → concatMapM

extra is already a build-depend. Use it:

-- current
tps <- concat <$> mapM processGroup (HM.toList byTopic)
-- preferred
tps <- concatMapM processGroup (HM.toList byTopic)

Telemetry.hs — BISECT_CAP_MAY_NOT_ISOLATE fires at logAttention

logAttention is alert-level. This fires on every call when V.length records > 1024 — a capacity advisory, not an error. Use logInfo or logDebug so on-call is not paged for a sizing note during normal high-volume ingestion.

Telemetry.hs — rawInsert: case chunks of → pattern guard

ViewPatterns and LambdaCase are already enabled. A pattern guard on the single-chunk arm is cleaner and shorter:

rawInsert pairs
  | [single] <- chunks = Hasql.session (HSession.statement () (chunkStmt single))
  | otherwise = Hasql.transaction TxS.ReadCommitted TxS.Write $ Relude.sum <$> traverse (Tx.statement () . chunkStmt) chunks
  where
    chunkSize = 700
    chunks = unfoldr (\v -> if V.null v then Nothing else Just (V.splitAt chunkSize v)) (V.map snd pairs)
    chunkStmt c = dynamicallyParameterized (otelInsertHeader <> mconcat (intersperse ", " (V.toList c))) D.rowsAffected True

What looks good

• isTransientException export eliminates all four maybe False isTransientHasqlError . fromException repetitions.
• sequence_ results in publishToDeadLetterQueue is correct and idiomatic — Either Text short-circuits on the first Left via its Monad instance.
• tpsFor: Map.fromListWith max + list comprehension + the >= 0 sentinel guard is exactly right.
• Bisection depth cap of 10 (2^10 = 1024) matches the stated contract.
• Removal of isUnrecoverableError (string-match heuristic) in favour of typed exception classification is a strict correctness improvement — unknown errors now route to DLQ rather than being silently swallowed.
• TF-strict semantics (throwIO eTf on TF failure, log-and-continue on PG failure) match the stated intent and phase-out plan.

The three items above (concatMapM, log level, pattern guard) are the main wins worth applying before merge.

Code Review — ingestion: TF-strict commit, per-partition offsets, row-level bisection

The durability semantics are sound, the tradeoffs are well-documented, and the overall design is a significant correctness improvement. A few issues worth addressing:

Bug: cabal.project.freeze index-state rolled back

-index-state: hackage.haskell.org 2026-05-31T00:00:00Z
+index-state: hackage.haskell.org 2026-05-30T16:55:48Z

Going backward in time means cabal may resolve to older package versions than what was pinned before. Looks like an accidental local-dev artifact — should be bumped forward to today or left at the previous value.

BISECT_CAP_MAY_NOT_ISOLATE should be logAttention, not logInfo

-- Telemetry.hs:1025
$ Log.logInfo "BISECT_CAP_MAY_NOT_ISOLATE"

When this fires, bisection cannot isolate a poison row, which means the partition will stall on repeat redeliveries. BISECT_DEPTH_EXHAUSTED (which is the consequence) uses logAttention — the early warning should too so on-call can act before the stall rather than after.

TF errors are double-logged

TIMEFUSION_WRITE_FAILED fires inside tfThread via whenLeft_, then throwIO eTf propagates out of bulkInsertOtelLogsAndSpansTF and is caught by tryAny in kafkaService, producing a second kafkaService: error processing batch entry for the same event. On-call ends up with two separate alert lines for one failure.

Options: drop the inner whenLeft_ log and let kafkaService log it with full context, or keep the inner log but suppress the rethrow path from reaching kafkaService's catch (harder). The former is simpler and eliminates the duplication.

isHasql inline check — consider a companion export

Telemetry.hs:878 still open-codes:

let isHasql = isJust (fromException @Hasql.HasqlException e)

This checks whether e is any HasqlException (infra vs. programmer-bug triage), which is distinct from isTransientException (transient vs. non-transient within Hasql). Since isTransientException was exported precisely to collapse this pattern family, a sibling isHasqlException :: SomeException -> Bool export would finish the job:

isHasqlException :: SomeException -> Bool
isHasqlException = isJust . fromException @HasqlException

One-liner in Data.Effectful.Hasql; the tfThread site collapses from two lines to one.

Minor: DLQ Left _ discards the error text at the call sites

Both pubsubService and kafkaService silently ignore the Left value from publishToDeadLetterQueue:

Left _ -> pure []

The error is already logged inside publishToDeadLetterQueue, so nothing is lost operationally. But for readers it's easy to assume the _ means "error doesn't matter" rather than "already logged in the helper." A brief inline comment (-- logged inside publishToDeadLetterQueue) would prevent that misread.

Positives worth noting

• tpsFor — Map.fromListWith max + list comprehension is exactly the right shape. The >= 0 sentinel guard is well-explained.
• summaryDocHash — swapping TE.decodeUtf8 . BSL.toStrict for Relude's polymorphic decodeUtf8 is clean; unused imports go with it.
• isTransientException export reduces 4 call sites from maybe False isTransientHasqlError . fromException to a single named predicate — good consolidation.
• sequence_ results in publishToDeadLetterQueue is idiomatic and correct (Either's >>= short-circuits on first Left; all IO effects have already run inside forM).

Code Review

Overall this is a solid durability fix. The per-partition commit strategy, isTransientException extraction, and DLQ gating on Either are all correct and improve the invariants meaningfully. A few issues worth addressing before merge:

Bug: type error in Worker.hs:summaryDocHash

-- Old (correct):
summaryDocHash = toXXHash . TE.decodeUtf8 . BSL.toStrict . AE.encode

-- New:
summaryDocHash = toXXHash . decodeUtf8 . AE.encode

aeson's encode returns Lazy.ByteString. Relude's decodeUtf8 (re-exported from Data.Text.Encoding) takes strict ByteString. This is a type error unless something else brings a lazy-aware decodeUtf8 into scope. The BSL.toStrict step needs to stay (or use Data.Text.Lazy.Encoding.decodeUtf8 + keep Data.Text.Lazy in scope).

Bug: bisection left-half commit + transient right-half → PG duplicates on retry

In bulkInsertOtelLogsAndSpans.go, when bisection begins (non-transient batch failure), rawInsert l and rawInsert r are separate transactions:

(+) <$> go (d - 1) l <*> go (d - 1) r

If go (d-1) l succeeds (left half committed to PG) and go (d-1) r then throws transiently, the exception propagates up through retryTimefusion, which retries the whole original batch via bulkInsertOtelLogsAndSpans records. The first thing that runs is rawInsert pairs (full batch) — if that now succeeds, the left-half rows are inserted again without deduplication.

This requires a specific failure sequence (non-transient full-batch → bisect → left succeeds → right transient), and PG is being phased out, so severity is low. Worth a comment or a WARN log if this path is ever taken (you can detect it by tracking whether any sub-batch has already succeeded before a transient escapes).

Correctness concern: right half silently skipped when left is depth-exhausted

-- <*> is sequential in Eff es: a throw from the left half
-- short-circuits and skips the right half.

The comment notes this but understates the consequence: if the left half hits BISECT_DEPTH_EXHAUSTED and throws, the right half (which might contain entirely good rows) never runs. Both halves then get treated as poison at the Kafka level and go to DLQ together. This could be surprising during outages where only a few rows are bad. Consider using liftA2 (+) (tryAny (go (d-1) l)) (tryAny (go (d-1) r)) with explicit re-throw logic to at least attempt the right half independently, or document this trade-off more explicitly.

Minor: sequence_ comment is slightly misleading

-- sequence_ over [Either Text ()] yields the FIRST Left
pure $ sequence_ results

sequence_ returns m (), not m a. It works here only because the element type is already (), making Either Text () the result. A clearer formulation:

pure $ sequenceA_ results  -- or: void (sequenceA results)

Or just annotate: sequence_ results :: Either Text (). The current comment implies sequence_ preserves the value, which it normally doesn't.

Minor: tpsFor — offset overflow sentinel

K.unOffset r.crOffset + 1

If crOffset is maxBound :: Int64, this overflows. Practically impossible but cheap to guard:

[(r.crPartition, K.unOffset r.crOffset + 1) | r <- toList neRecords, let o = K.unOffset r.crOffset, o >= 0, o < maxBound]

Positive observations

• isTransientException extraction is a clean DRY improvement — collapsing 4 call sites of maybe False isTransientHasqlError . fromException is exactly the right move.
• Replacing isUnrecoverableError (brittle string matching) with structured isTransientException is strictly better; unknown errors going to DLQ rather than silently stalling is correct.
• Map.fromListWith max in tpsFor is idiomatic and correct.
• processGroup factoring in kafkaService improves readability without adding indirection.
• void $ tryAny (batch (V.singleton r)) in SchemaCatalog.hs is a nice clean-up.

Code Review

The durability fix is well-motivated and the overall approach is sound. A few findings on correctness, succinctness, and missed package opportunities:

Potential Bug

cabal.project.freeze index-state rolled backwards

-index-state: hackage.haskell.org 2026-05-31T00:00:00Z
+index-state: hackage.haskell.org 2026-05-30T16:55:48Z

The timestamp moves backwards by ~7 hours, making packages published in that window invisible to Cabal resolution. Was this intentional (pinning to a known-good snapshot), or an accidental revert?

Succinctness / Package Improvements

1. sequence_ vs. lefts + listToMaybe in publishToDeadLetterQueue

The PR description says "Result assembly uses partitionEithers + listToMaybe" but the code uses sequence_:

-- actual code
pure $ sequence_ results

sequence_ works here (short-circuits on the first Left via Either's Applicative), but it relies on non-obvious instance semantics. The description's stated idiom, using imports already in scope at line 236, is more explicit about intent:

pure . maybe (Right ()) Left . listToMaybe . lefts $ results

Alternatively partitionEithers results & \(errs, _) -> pure $ maybe (Right ()) Left (listToMaybe errs). Either way, the PR description and code should agree.

2. isJust (fromException @Hasql.HasqlException e) still open-coded

The PR correctly introduces isTransientException to collapse four transient-check call sites. But the pattern isJust (fromException @Hasql.HasqlException e) — checking whether an exception is any HasqlException (for infra vs programmer-error classification) — is still open-coded in at least three places:

• src/Models/Telemetry/Telemetry.hs:878
• src/Opentelemetry/OtlpServer.hs:1686
• src/BackgroundJobs.hs:1426

A companion export alongside isTransientException would be:

isHasqlException :: SomeException -> Bool
isHasqlException = isJust . fromException @HasqlException

Not a blocker — different semantics from isTransientException — but the same dedup rationale applies.

Minor

rawInsert guard — silent empty-chunks case

rawInsert pairs
  | [single] <- chunks = ...
  | otherwise           = ...  -- handles both multi-chunk AND empty

The otherwise arm silently handles the empty chunks case (sum over empty traverse → 0). The invariant "rawInsert is never called with null pairs" holds because go is only entered when not (V.null pairs), but it's not locally obvious. The original case chunks of { [single] -> ...; _ -> ... } was equally implicit. A brief note or assert would help future readers.

Positive Highlights

• Per-partition commits via tpsFor + commitPartitionsOffsets are a clear upgrade over commitAllOffsets; the Map.fromListWith max offset computation and sentinel guard (>= 0) are correct.
• Bisection in bulkInsertOtelLogsAndSpans is well-structured; depth cap, BISECT_CAP_MAY_NOT_ISOLATE warning, and $> 0 on poison-row drop are all tight.
• summaryDocHash: leveraging Relude's ConvertUtf8 Text LByteString to drop BSL.toStrict . TE.decodeUtf8 is exactly the kind of cleanup this codebase values.
• void $ tryAny ... replacing >>= either (const pass) (const pass) — nice.
• Removing isUnrecoverableError string matching in favor of type-based dispatch is a genuine reliability improvement.
• TF-strict semantics and the BOTH_WRITES_FAILED split are clearly documented and the pgThread wrapping tryAny correctly prevents a PG exception from killing the Ki scope before the TF fork completes.