feat(source): v1 hardening — tx-Term contract, dependency isolation, test-gap closure

.github/workflows/source.yml

@@ -103,19 +103,21 @@ jobs:
           awk -v p="$pct" -v t="$THRESHOLD" 'BEGIN { exit (p+0 < t+0) ? 1 : 0 }' \
             || { echo "::error::coverage ${pct}% is below the ${THRESHOLD}% threshold"; exit 1; }
 
-  # Integration / E2E leg, behind the `integration` build tag, run with
-  # GOWORK=off. The in-workspace source modules run end-to-end; the SDK-backed
-  # source modules drive a real broker via testcontainers (Docker is available on
-  # GitHub-hosted ubuntu runners) and skip cleanly if a daemon is ever absent.
-  # Kept off the default test matrix so the hermetic checks stay fast.
+  # Integration / E2E leg, run with GOWORK=off. The in-workspace source modules
+  # run end-to-end; the SDK-backed source modules drive a real broker via
+  # testcontainers (Docker is available on GitHub-hosted ubuntu runners) and skip
+  # cleanly if a daemon is ever absent. source/kafka's RedPanda leg lives in the
+  # test-only nested module source/kafka/integration so its testcontainers,
+  # redpanda, and kadm dependencies never enter source/kafka's go.mod. Kept off
+  # the default test matrix so the hermetic checks stay fast.
   integration:
     strategy:
       fail-fast: false
       matrix:
         module:
           - source
           - source/statemachine
-          - source/kafka
+          - source/kafka/integration
           - source/jetstream
           - source/redis
           - source/cloudevents

magefiles/magefile.go

@@ -445,8 +445,16 @@ var integrationModules = func() []string {
 	mods := []string{"sink/file", "sink/http", "sink/prometheus", "sink/slog"}
 	mods = append(mods, sinkDestinations...)
 	mods = append(mods, "source", "source/statemachine")
-	mods = append(mods, sourceDestinations...)
-	return mods
+	for _, mod := range sourceDestinations {
+		if mod == "source/kafka" {
+			// The RedPanda leg lives in its own test-only module so its
+			// testcontainers/redpanda/kadm deps stay out of source/kafka's
+			// go.mod; appended below under its nested path.
+			continue
+		}
+		mods = append(mods, mod)
+	}
+	return append(mods, "source/kafka/integration")
 }()
 
 // Integration runs the //go:build integration leg for every sink and source

source/hopper.go

@@ -209,7 +209,7 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error {
 			select {
 			case inFlight <- struct{}{}:
 			case <-runCtx.Done():
-				return hp.exitErr(ctx, runErr)
+				return hp.exitErr(runErr)
 			}
 		}
 
@@ -219,10 +219,10 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error {
 				<-inFlight // release the slot we reserved but did not use
 			}
 			if errors.Is(err, ErrDrained) || errors.Is(err, context.Canceled) || runCtx.Err() != nil {
-				return hp.exitErr(ctx, runErr)
+				return hp.exitErr(runErr)
 			}
 			errOnce.Do(func() { runErr = err })
-			return hp.exitErr(ctx, runErr)
+			return hp.exitErr(runErr)
 		}
 
 		hp.received.Add(runCtx, 1)
@@ -236,7 +236,7 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error {
 			if inFlight != nil {
 				<-inFlight
 			}
-			return hp.exitErr(ctx, runErr)
+			return hp.exitErr(runErr)
 		}
 
 		workWG.Add(1)
@@ -250,17 +250,15 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error {
 			if inFlight != nil {
 				<-inFlight
 			}
-			return hp.exitErr(ctx, runErr)
+			return hp.exitErr(runErr)
 		}
 	}
 }
 
 // exitErr maps a run's terminal error onto the public contract: a clean drain
 // ([ErrDrained]), a context cancellation, and a Close are all graceful and
-// return nil; only a genuine fetch error propagates. The parent context is
-// accepted to document that its cancellation is the expected, non-error
-// shutdown path.
-func (hp *Hopper) exitErr(_ context.Context, runErr error) error {
+// return nil; only a genuine fetch error propagates.
+func (hp *Hopper) exitErr(runErr error) error {
 	return runErr
 }
 

source/hopper_test.go

@@ -591,3 +591,127 @@ func TestHopper_RedeliveryStormBoundedResources(t *testing.T) {
 		time.Sleep(10 * time.Millisecond)
 	}
 }
+
+// settleBlockingSub mimics a backend whose Settle honors ctx and can be slow —
+// a Kafka commit or JetStream ack mid-round-trip when shutdown lands. Next
+// yields the queued messages, then blocks on ctx; after Close it never yields
+// again. Settle blocks until the context is canceled, then completes exactly
+// once, so the test can prove the engine neither strands nor double-settles
+// in-flight work across a Close+cancel race.
+type settleBlockingSub struct {
+	mu               sync.Mutex
+	msgs             []source.Message
+	closed           bool
+	nextCalls        int
+	yieldsAfterClose int
+
+	settleStarted chan struct{}
+	startOnce     sync.Once
+	settleCalls   atomic.Int32
+}
+
+func newSettleBlockingSub(msgs ...source.Message) *settleBlockingSub {
+	return &settleBlockingSub{msgs: msgs, settleStarted: make(chan struct{})}
+}
+
+func (s *settleBlockingSub) Next(ctx context.Context) (source.Message, error) {
+	s.mu.Lock()
+	s.nextCalls++
+	closed := s.closed
+	if len(s.msgs) > 0 {
+		m := s.msgs[0]
+		s.msgs = s.msgs[1:]
+		if closed {
+			s.yieldsAfterClose++
+		}
+		s.mu.Unlock()
+		return m, nil
+	}
+	s.mu.Unlock()
+	if closed {
+		return nil, source.ErrDrained
+	}
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (s *settleBlockingSub) Settle(ctx context.Context, _ source.Message, _ source.Result) error {
+	s.settleCalls.Add(1)
+	s.startOnce.Do(func() { close(s.settleStarted) })
+	<-ctx.Done() // the backend settle only completes when its context ends
+	return nil
+}
+
+func (s *settleBlockingSub) Close() error {
+	s.mu.Lock()
+	s.closed = true
+	s.mu.Unlock()
+	return nil
+}
+
+func (s *settleBlockingSub) yieldsAfterCloseCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.yieldsAfterClose
+}
+
+// TestHopper_DrainOnCancelWithMidFlightSettle pins the shutdown contract when a
+// backend settle is mid-flight: a Close and a context cancel racing must leave
+// exactly one settle (no double-settle), yield nothing new after Close, strand
+// no goroutines, and let Run return.
+func TestHopper_DrainOnCancelWithMidFlightSettle(t *testing.T) {
+	t.Parallel()
+
+	sub := newSettleBlockingSub(testMsg{key: []byte("k"), value: []byte("v")})
+	hp := source.New(source.WithConcurrency(1))
+	t.Cleanup(func() { _ = hp.Close() })
+
+	baseline := runtime.NumGoroutine()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		done <- hp.Run(ctx, sub, func(context.Context, source.Message) source.Result {
+			return source.Ack()
+		})
+	}()
+
+	select {
+	case <-sub.settleStarted:
+	case <-time.After(3 * time.Second):
+		t.Fatal("message never reached its (blocking) settle")
+	}
+
+	// Race graceful Close against cancellation, like an operator shutdown and
+	// a deadline expiring at once.
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() { defer wg.Done(); _ = sub.Close() }()
+	go func() { defer wg.Done(); cancel() }()
+	wg.Wait()
+
+	select {
+	case err := <-done:
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, source.ErrDrained) {
+			t.Fatalf("Run = %v, want nil, context.Canceled, or ErrDrained", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("Run did not return after the Close+cancel race")
+	}
+
+	if got := sub.settleCalls.Load(); got != 1 {
+		t.Fatalf("settle calls = %d, want exactly 1 (no double-settle)", got)
+	}
+	if got := sub.yieldsAfterCloseCount(); got != 0 {
+		t.Fatalf("Next yielded %d messages after Close, want 0", got)
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for runtime.NumGoroutine() > baseline+2 {
+		if time.Now().After(deadline) {
+			t.Fatalf("goroutines after run = %d, want within +2 of baseline %d (stranded in-flight)",
+				runtime.NumGoroutine(), baseline)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}

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      |
+| `Term`        | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`; route poison through `Begin` instead)      |
 | `InProgress`  | no-op (Kafka has no per-message ack deadline)               |
 | `Manual`      | no-op (the handler settled via `Message.As` + the client)   |
 
@@ -55,6 +55,15 @@ 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.
 
+### Divergence: Term inside a transaction
+
+On a subscription built with `WithTransactional`, settling `Term` directly is
+rejected with `ErrTermInsideTransaction`: a DLQ produce outside the open EOS
+session would not be atomic with the consumed offset. Inside
+`Transactional.Begin`, produce the poison record to the dead-letter topic via
+the handed `source.Tx` and return nil — the DLQ record and the input offset
+then commit as one atomic unit.
+
 ## Capabilities
 
 The subscription satisfies these optional `source` capability interfaces,

source/kafka/adapter_internal_test.go

@@ -747,3 +747,61 @@ func TestSubscribeBuildsClientAndAs(t *testing.T) {
 type stringCursor string
 
 func (c stringCursor) String() string { return string(c) }
+
+// TestSettleTermOnTransactionalSubscriptionRejected pins the P2-9 decision: on
+// a transactional subscription a direct Term would produce the DLQ record
+// outside the open EOS session, breaking atomicity with the offset mark, so it
+// is rejected with ErrTermInsideTransaction and has no side effects.
+func TestSettleTermOnTransactionalSubscriptionRejected(t *testing.T) {
+	t.Parallel()
+
+	ft := &fakeTransactor{committed: true}
+	sub, fp := newTxSub(ft)
+	m := newMessage(rec("orders", 0, 9, "A-1", "poison"))
+
+	err := sub.Settle(context.Background(), m, source.Term(errors.New("poison")))
+	if !errors.Is(err, ErrTermInsideTransaction) {
+		t.Fatalf("Settle(Term) on tx subscription = %v, want ErrTermInsideTransaction", err)
+	}
+	if got := fp.markedCount(); got != 0 {
+		t.Errorf("marked = %d, want 0 (rejected settle must not advance offsets)", got)
+	}
+	if got := fp.producedCount(); got != 0 {
+		t.Errorf("produced = %d, want 0 (no DLQ write outside the transaction)", got)
+	}
+	if len(ft.calls) != 0 {
+		t.Errorf("transact calls = %v, want none", ft.calls)
+	}
+}
+
+// TestBeginDeadLettersThroughTransactionOnPoison pins the supported EOS DLQ
+// pattern for poison messages: inside Begin's fn, produce the rejected record
+// to the dead-letter topic via the handed Tx and return nil, so the DLQ
+// record and the consumed offset commit as one atomic unit.
+func TestBeginDeadLettersThroughTransactionOnPoison(t *testing.T) {
+	t.Parallel()
+
+	ft := &fakeTransactor{committed: true}
+	sub, fp := newTxSub(ft)
+	m := newMessage(rec("orders", 0, 11, "A-1", "poison"))
+
+	err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error {
+		return tx.Produce(ctx, source.ProducedRecord{
+			Topic: "orders.DLQ",
+			Key:   []byte("A-1"),
+			Value: []byte("poison"),
+		})
+	})
+	if err != nil {
+		t.Fatalf("Begin() error = %v, want nil", err)
+	}
+	if len(ft.produced) != 1 || len(ft.produced[0]) != 1 || ft.produced[0][0].Topic != "orders.DLQ" {
+		t.Fatalf("produced = %#v, want one record on orders.DLQ", ft.produced)
+	}
+	if got := fp.markedCount(); got != 1 {
+		t.Errorf("marked = %d, want 1 (consumed offset committed atomically with the DLQ record)", got)
+	}
+	if got, want := ft.calls, []string{"begin", "produce", "end-commit"}; !equalStrings(got, want) {
+		t.Errorf("call order = %v, want %v", got, want)
+	}
+}

source/kafka/capability.go

@@ -432,6 +432,12 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader {
 // without a transaction. A rebalance during the transaction fences the producer,
 // so End reports the transaction did not commit and the work is retried after
 // 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.
 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)