docs(source): resolve the second gate pass's remaining contract-doc findings

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
 }

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

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)

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.