From ef8308f1c81cc15d1d2ab85204ef41c35f963b54 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 23 Aug 2026 03:10:45 -0400 Subject: [PATCH] =?UTF-8?q?docs(source):=20resolve=20the=20third=20gate=20?= =?UTF-8?q?pass's=20findings=20=E2=80=94=20scoping=20honesty=20+=20poison-?= =?UTF-8?q?loop=20and=20panic=20semantics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 (doc-consistency): doc.go no longer promises 'exactly within a session' redelivery for offset-riding backends; it now scopes in-session redelivery to each adapter's documented constraints (naming Kafka's concurrent-commit caveat explicitly), matching handler.go and the kafka README. P2s: Nak-after-Ack outcome specified (does not unmark; persisted position stays past the record; an in-session re-seek may still redeliver once); foreign-settle detection boundary spelled out (sibling-subscription Kafka records apply as if delivered); missing-DLQ-topic steady state documented on the README setup notes (ErrNoDLQTopic per settle, record unmarked, partition stalls until configured); Begin documents that a panic inside fn is not recovered and leaves the transaction open until End, session close, or a rebalance fence. --- source/doc.go | 4 +++- source/inlet.go | 11 +++++++---- source/kafka/README.md | 5 +++++ source/kafka/capability.go | 5 ++++- 4 files changed, 19 insertions(+), 6 deletions(-) diff --git a/source/doc.go b/source/doc.go index 83bb96b..25ad95f 100644 --- a/source/doc.go +++ b/source/doc.go @@ -30,7 +30,9 @@ // restarts and rebalances, backends whose redelivery rides persisted offsets // (Kafka) resume from the last committed position, which a concurrently // committed higher offset can advance past a nacked record; those backends -// therefore redeliver exactly within a session and best-effort across restarts, +// therefore redeliver within the live session subject to each adapter's +// documented constraints (Kafka: a concurrent commit can pass a nacked record +// before its re-seek lands) and best-effort across restarts, // and each adapter documents the precise semantics in its own module. A handler // returns a [Result] — [Ack], [Nak], [Term], [InProgress], or [Manual] — and // the Hopper applies it to the backend. Backends differ (Kafka commits offsets diff --git a/source/inlet.go b/source/inlet.go index 818db84..aa61e0e 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -64,11 +64,14 @@ type Subscription interface { // every call is applied and recorded in arrival order, and in-flight // accounting never goes negative, so drain semantics are unaffected. A // later decision supersedes an earlier one for future commits — an Ack - // after a Nak marks past the record — though a re-seek already issued may - // still yield one final redelivery (at-least-once). Settling a message this + // after a Nak marks past the record, though a re-seek already issued may + // still yield one final redelivery (at-least-once); a Nak after an Ack does + // not unmark, so the persisted position stays past the record while any + // in-session re-seek may still redeliver it once. Settling a message this // subscription never delivered, or after drain completed, is - // backend-defined: where detectable it is rejected (Kafka rejects non-Kafka - // messages); otherwise it applies as if delivered. + // backend-defined: where detectable it is rejected (Kafka rejects messages + // that are not Kafka records; a record from a sibling subscription of the + // same backend is applied as if delivered). Settle(ctx context.Context, m Message, r Result) error // Close begins a graceful drain: from then on Next stops yielding fresh // messages — only messages already handed out or buffered before Close may diff --git a/source/kafka/README.md b/source/kafka/README.md index 485b0ea..d23a2d7 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -122,6 +122,11 @@ track franz-go releases. ## Setup notes +- Configure `WithDLQTopic` before processing poison-prone streams. Without it, + every `Term` fails with `ErrNoDLQTopic` and the record stays unmarked — a + poison record redelivers on every fetch, stalling its partition until the + dead-letter topic is configured or the DLQ produce recovers. + - The dead-letter topic must exist before the first `Term` settles: the adapter's DLQ producer does not enable topic auto-creation. Create it up front (as the integration suite does with `CreateTopics`). diff --git a/source/kafka/capability.go b/source/kafka/capability.go index b359696..e1af782 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -428,7 +428,10 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // for m. A direct Settle for m during an open Begin, or after Begin returned, // is a caller error with undefined results — it is not detected or coordinated // with the transaction: at best it duplicates an already-committed mark -// (harmless at-least-once), at worst it races End's commit/abort. +// (harmless at-least-once), at worst it races End's commit/abort. fn must not +// panic: a panic is not recovered; it propagates to the caller of Begin, +// leaving the transaction open until a later End, the session closes, or a +// rebalance fences the producer (which aborts it). // // It is available only when the inlet was built with [WithTransactional]; // otherwise it reports the capability is absent rather than silently running fn