From a9d912efc09140f08c99b245a3d50a43c4de4fc2 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 23 Aug 2026 00:55:50 -0400 Subject: [PATCH] docs(source): close contract-doc gaps found by the adversarial review gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer-panel findings, all documentation-level (implementation verified conforming by executed experiments): - P1: DLQ produce-failure semantics were unspecified. Documented on the Term row: a failed dead-letter produce returns the error from Settle and leaves the record unmarked, so it is redelivered — nothing is silently lost. - handler.go ActionNak no longer states an unconditional in-session redelivery guarantee; it now defers to each adapter's documented constraints (Kafka's concurrent-commit caveat), matching source/doc.go's scoping. - inlet.go Next: ctx cancellation during the drain wait takes precedence over ErrDrained; the buffered-yield clause now covers Batched.NextBatch records fetched but not yet returned; Settle documents duplicate-settle safety. - kafka capability.go Begin: repeated Term on a transactional subscription leaves the record unmarked (hot-loop consequence stated); mixing a direct Settle into an open Begin documented as a programming error. - kafka README: NakAfter(d) pause interaction with BlockRebalanceOnPoll clarified. --- source/handler.go | 13 +++++++------ source/inlet.go | 18 ++++++++++++------ source/kafka/README.md | 6 ++++-- source/kafka/capability.go | 13 +++++++++---- 4 files changed, 32 insertions(+), 18 deletions(-) diff --git a/source/handler.go b/source/handler.go index a788765..0235e75 100644 --- a/source/handler.go +++ b/source/handler.go @@ -44,12 +44,13 @@ const ( ActionAck Action = iota // ActionNak asks for redelivery: the message failed transiently and should // be tried again (JetStream naks natively with optional delay; Kafka pauses - // and re-seeks the record's partition so it is fetched again). Redelivery - // is guaranteed within a live subscription; across process restarts or - // rebalances, backends whose redelivery rides persisted offsets (Kafka) - // honor the last committed position, so a concurrently committed higher - // offset can pass a nacked record — each adapter documents its exact - // semantics. Result.Requeue is an optional backoff delay. + // and re-seeks the record's partition so it is fetched again). Within the + // constraints each adapter documents (Kafka's concurrent-commit caveat is + // described in its README), redelivery happens in the live subscription; + // across process restarts or rebalances, backends whose redelivery rides + // persisted offsets (Kafka) honor the last committed position, so a + // concurrently committed higher offset can pass a nacked record. + // Result.Requeue is an optional backoff delay. ActionNak // ActionTerm rejects the message permanently: it must not be redelivered // (JetStream Term; Kafka routes it to a dead-letter topic, then commits). diff --git a/source/inlet.go b/source/inlet.go index a66acdd..d9b4244 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -48,16 +48,22 @@ type Inlet interface { // goroutines and must be safe for that. type Subscription interface { // Next returns the next message. It blocks until one is available, returns - // ctx.Err() if ctx is canceled, or returns ErrDrained once the subscription - // has been closed and all delivered messages settled. After Close, a - // backend must not deliver new messages: only messages already returned to - // the buffer before Close may still be yielded, and once none remain, - // Next blocks until in-flight settles finish and then reports ErrDrained. + // ctx.Err() if ctx is canceled — including while it waits for in-flight + // settles during a drain, which takes precedence over ErrDrained — or + // returns ErrDrained once the subscription has been closed and all + // delivered messages settled. After Close, a backend must not deliver new + // messages: only messages already handed out or buffered before Close may + // still be yielded (for [Batched.NextBatch], this includes records fetched + // but not yet returned), and once none remain, Next blocks until in-flight + // settles finish and then reports ErrDrained. Next(ctx context.Context) (Message, error) // Settle applies a handler [Result] to a message previously returned by Next: // 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. + // 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. 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 diff --git a/source/kafka/README.md b/source/kafka/README.md index 9d6ed67..17acf73 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) | +| `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 | | `InProgress` | no-op (Kafka has no per-message ack deadline) | | `Manual` | no-op (the handler settled via `Message.As` + the client) | @@ -90,7 +90,9 @@ do expose it: `BlockRebalanceOnPoll` gives the engine a safe processing window: a rebalance cannot move partitions mid-batch; the subscription releases the rebalance only -between fetches. +between fetches. A `NakAfter(d)` pause holds that window for at most `d` per +event: the partition pause blocks further fetches of that partition, but the +rebalance itself proceeds at the next release boundary. ## Cold start (initial start offset) diff --git a/source/kafka/capability.go b/source/kafka/capability.go index 19ad33a..03c405d 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -434,10 +434,15 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // reassignment. // // Poison handling on a transactional subscription flows through Begin too: -// settling [source.Term] directly is rejected with [ErrTermInsideTransaction]. -// 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. +// settling [source.Term] directly is rejected with [ErrTermInsideTransaction]; +// every such attempt leaves the record unmarked, so a handler retrying Term in +// 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. 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)