From dfda1c430ffc9e46732612c35964ac6669cac718 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 23 Aug 2026 01:26:26 -0400 Subject: [PATCH] docs(source): resolve the second gate pass's remaining contract-doc findings P1: the Term row (README) and package-doc ack model now state that ErrTermInsideTransaction leaves the record unmarked and redelivered, so a hot-looping poison handler's failure mode is explicit on every surface that mentions the rejection. P2s: duplicate-settle semantics now specify arrival-order composition (a later Ack supersedes an earlier Nak for future commits; an already-issued re-seek may still yield one final at-least-once redelivery); settling a never-delivered or post-drain message is defined as backend-defined with Kafka's detection noted; violating Begin's only-settle-path rule is explicitly undefined behavior; and the Close doc wording no longer tensions with Next's buffered-yield clause. --- source/inlet.go | 18 ++++++++++++------ source/kafka/README.md | 2 +- source/kafka/capability.go | 9 +++++---- source/kafka/kafka.go | 7 ++++--- 4 files changed, 22 insertions(+), 14 deletions(-) diff --git a/source/inlet.go b/source/inlet.go index d9b4244..818db84 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -61,12 +61,18 @@ type Subscription interface { // ack/commit, schedule redelivery, route to dead-letter, or extend the // deadline, per Result.Action. It is the single point where a delivery // decision reaches the backend. Settling a message more than once is safe: - // every call is applied and recorded (a duplicate mark is idempotent at - // every backend), and in-flight accounting never goes negative, so drain - // semantics are unaffected. + // 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 + // 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. Settle(ctx context.Context, m Message, r Result) error - // Close begins a graceful drain: Next stops yielding new messages, and once - // in-flight messages are settled, Next returns ErrDrained. Close is - // idempotent. + // Close begins a graceful drain: from then on Next stops yielding fresh + // messages — only messages already handed out or buffered before Close may + // still be yielded — and once in-flight messages are settled, Next returns + // ErrDrained. Close is idempotent. Close() error } diff --git a/source/kafka/README.md b/source/kafka/README.md index 17acf73..485b0ea 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -35,7 +35,7 @@ and the marked offsets are committed on graceful drain and on rebalance | `Ack` | mark the record for commit (commit-after-process) | | `Nak` | never mark; pause, re-seek to the record's offset, resume — the record is fetched again in this session | | `NakAfter(d)` | same as `Nak`, waiting out `d` between pause and re-seek (best-effort) | -| `Term` | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`; route poison through `Begin` instead). If the DLQ produce fails, `Settle` returns the error and the record is **not** marked — it is redelivered; nothing is silently lost | +| `Term` | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`, leaving the record unmarked so it is redelivered; route poison through `Begin` instead). If the DLQ produce fails, `Settle` returns the error and the record is **not** marked — it is redelivered; nothing is silently lost | | `InProgress` | no-op (Kafka has no per-message ack deadline) | | `Manual` | no-op (the handler settled via `Message.As` + the client) | diff --git a/source/kafka/capability.go b/source/kafka/capability.go index 03c405d..b9901c1 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -439,10 +439,11 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // a loop simply sees the record again on its next fetch — route it through // Begin instead. Inside fn, produce the rejected record to the dead-letter // topic via the handed [source.Tx] and return nil, so the DLQ write and the -// consumed offset commit atomically. Begin is the only settle path for m while -// it runs: mixing a direct [Subscription.Settle] for m into an open Begin is a -// programming error; the direct settle marks independently and can double- -// commit or race the transaction's End. +// consumed offset commit atomically. Begin is the only settle path for m when +// used transactionally: mixing a direct [Subscription.Settle] for m into an +// open Begin is a caller error and its behavior is undefined — the direct +// settle marks independently of the transaction, so it can double-advance the +// partition or race End's commit/abort. func (s *subscription) Begin(ctx context.Context, m source.Message, fn func(ctx context.Context, tx source.Tx) error) error { if s.transactSess == nil { return fmt.Errorf("source/kafka: transactional: %w", errNotTransactional) diff --git a/source/kafka/kafka.go b/source/kafka/kafka.go index 460d9fa..a14d930 100644 --- a/source/kafka/kafka.go +++ b/source/kafka/kafka.go @@ -24,9 +24,10 @@ // not an in-session one. // - Term produces the record to the configured dead-letter topic, then marks // it for commit so it is not re-read — except on a transactional -// subscription, where Term reports [ErrTermInsideTransaction]: route poison -// through Begin's transaction instead, producing to the dead-letter topic -// via the handed [source.Tx] so DLQ write and offset commit are atomic. +// subscription, where Term reports [ErrTermInsideTransaction] and leaves +// the record unmarked, so it is redelivered: route poison through Begin's +// transaction instead, producing to the dead-letter topic via the handed +// [source.Tx] so DLQ write and offset commit are atomic. // - InProgress is a no-op: Kafka has no per-message ack deadline to extend. // - Manual is a no-op: the handler settled the record itself through // [source.Message.As] and the underlying *kgo.Client.