fix(source/kafka): honor drain contract after Close; make Nak redeliver in-session

source/doc.go

@@ -23,15 +23,22 @@
 //
 // # Contract
 //
-// Delivery is at-least-once by default: a message is acked only after its
-// handler reports success, never before processing. A handler returns a
-// [Result] — [Ack], [Nak], [Term], [InProgress], or [Manual] — and the Hopper
-// applies it to the backend. Backends differ (Kafka commits offsets per
-// partition; JetStream acks per message), so capabilities a backend may or may
-// not have — replay, consumer groups, transactions — are optional interfaces
-// discovered by type assertion ([Seekable], [ConsumerGroups], [Transactional],
-// …) rather than a lowest-common-denominator API that lies about what a backend
-// can do.
+// Delivery is at-least-once within a live subscription: a message is acked only
+// after its handler reports success, never before processing, and a [Nak]
+// redelivers — JetStream naks natively (with optional delay), Kafka pauses and
+// re-seeks the record's partition so it is fetched again. Across process
+// 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,
+// 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
+// per partition; JetStream acks per message), so capabilities a backend may or
+// may not have — replay, consumer groups, transactions — are optional
+// interfaces discovered by type assertion ([Seekable], [ConsumerGroups],
+// [Transactional], …) rather than a lowest-common-denominator API that lies
+// about what a backend can do.
 //
 // # No forced dependencies
 //

source/handler.go

@@ -43,8 +43,13 @@ const (
 	// means "ack".
 	ActionAck Action = iota
 	// ActionNak asks for redelivery: the message failed transiently and should
-	// be tried again (JetStream Nak/NakWithDelay; Kafka declines to commit so the
-	// record is re-read). Result.Requeue is an optional backoff delay.
+	// 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.
 	ActionNak
 	// ActionTerm rejects the message permanently: it must not be redelivered
 	// (JetStream Term; Kafka routes it to a dead-letter topic, then commits).

source/hopper_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -499,3 +500,94 @@ func TestHopper_MaxInFlightBackpressure(t *testing.T) {
 	}
 	h.AssertSettled(total)
 }
+
+// stormSub is a minimal Subscription whose Settle requeues nacked messages, so
+// a handler that keeps Naking drives Hopper.run through a real redelivery loop
+// without a broker.
+type stormSub struct {
+	queue chan source.Message
+}
+
+func newStormSub(m source.Message) *stormSub {
+	q := make(chan source.Message, 8)
+	q <- m
+	return &stormSub{queue: q}
+}
+
+func (s *stormSub) Next(ctx context.Context) (source.Message, error) {
+	select {
+	case m := <-s.queue:
+		return m, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (s *stormSub) Settle(_ context.Context, m source.Message, r source.Result) error {
+	if r.Action == source.ActionNak {
+		s.queue <- m // redeliver
+	}
+	return nil
+}
+
+func (s *stormSub) Close() error { return nil }
+
+// TestHopper_RedeliveryStormBoundedResources drives 500 consecutive Naks of a
+// single-key message through a real Hopper.run redelivery loop and pins the
+// resource contract: Run terminates, the delivery loop really cycled (>= 501
+// deliveries for 500 naks + 1 ack), and every lane/fetch goroutine unwinds so
+// the goroutine count returns to its pre-run baseline.
+func TestHopper_RedeliveryStormBoundedResources(t *testing.T) {
+	t.Parallel()
+
+	sub := newStormSub(testMsg{key: []byte("k"), value: []byte("v")})
+	hp := source.New(source.WithConcurrency(4))
+	t.Cleanup(func() { _ = hp.Close() })
+
+	var deliveries atomic.Int64
+	acked := make(chan struct{})
+	var ackOnce sync.Once
+	handler := func(context.Context, source.Message) source.Result {
+		if deliveries.Add(1) <= 500 {
+			return source.Nak(errors.New("transient"))
+		}
+		ackOnce.Do(func() { close(acked) })
+		return source.Ack()
+	}
+
+	baseline := runtime.NumGoroutine()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- hp.Run(ctx, sub, handler) }()
+
+	select {
+	case <-acked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("handler never reached its first ack after the redelivery storm")
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil && !errors.Is(err, context.Canceled) {
+			t.Fatalf("Run = %v, want nil or context.Canceled", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+
+	if got := deliveries.Load(); got < 501 {
+		t.Fatalf("deliveries = %d, want >= 501 (500 naks + 1 ack through the redelivery loop)", got)
+	}
+
+	// Bounded resources: the fetch loop and lane goroutines must all unwind
+	// once Run returns, returning the process to its pre-run goroutine count.
+	deadline := time.Now().Add(2 * time.Second)
+	for runtime.NumGoroutine() > baseline+2 {
+		if time.Now().After(deadline) {
+			t.Fatalf("goroutines after run = %d, want back within +2 of baseline %d", runtime.NumGoroutine(), baseline)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}

source/inlet.go

@@ -49,7 +49,10 @@ type Inlet interface {
 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.
+	// 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.
 	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

source/kafka/README.md

@@ -33,19 +33,27 @@ and the marked offsets are committed on graceful drain and on rebalance
 | Result        | Kafka behavior                                              |
 | ------------- | ----------------------------------------------------------- |
 | `Ack`         | mark the record for commit (commit-after-process)           |
-| `Nak`         | do **not** mark; the record is re-read on restart/rebalance |
-| `NakAfter(d)` | best-effort: pause partition, wait `d`, re-seek, resume     |
+| `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      |
 | `InProgress`  | no-op (Kafka has no per-message ack deadline)               |
 | `Manual`      | no-op (the handler settled via `Message.As` + the client)   |
 
-### Divergence: Nak delay
-
-Kafka has no native per-message redelivery delay. `NakAfter(d)` is emulated by
-pausing the record's partition, waiting out `d` (or the context), re-seeking to
-the record's own offset, and resuming. A plain `Nak` simply declines to mark,
-so the record is re-delivered on the next restart or rebalance. This is the
-documented divergence from JetStream's native delayed nak.
+### Divergence: Nak delay and cross-restart redelivery
+
+Kafka has no native per-message redelivery. A `Nak` (plain or delayed) is
+emulated by pausing the record's partition, waiting out the requested delay
+(zero for a plain `Nak`), re-seeking the partition to the record's own offset,
+and resuming — so the record is redelivered deterministically **within the live
+subscription**. Costs to know about: the delay head-of-line-blocks the paused
+partition; records already fetched but not yet yielded are still delivered
+before the redelivered record; and concurrent settles on the same partition can
+commit past the nacked offset before the re-seek lands.
+
+Across process restarts and rebalances, redelivery rides committed offsets: a
+concurrently committed higher offset can advance past a nacked record, so
+cross-restart redelivery is **best-effort**, not guaranteed. This is the one
+divergence from JetStream's native nak semantics.
 
 ## Capabilities