docs(source): close contract-doc gaps found by the adversarial review gate

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).

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

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)
 

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)