feat: add Kafka exactly-once consume-process-produce

docs/src/content/docs/source/adapters.md

@@ -24,8 +24,9 @@ sub.Receive(ctx, handler)
 ```
 
 It does cooperative rebalance with drain-on-revoke, mark-commit-after-process,
-pause/resume backpressure, and seek/replay, and it is EOS-ready for transactional
-consume-to-produce. The shard is the partition; the
+pause/resume backpressure, and seek/replay, and it supports transactional
+consume-process-produce (see [Exactly-once (Kafka EOS)](#exactly-once-kafka-eos)
+below). The shard is the partition; the
 [high-water-mark commit subtlety](/crucible/source/concurrency/#the-kafka-high-water-mark-subtlety)
 is handled in the engine.
 
@@ -68,6 +69,63 @@ The divergences are documented, never papered over. A `Nak(delay)` is a real
 delayed redelivery on JetStream but is best-effort on Kafka (pause plus reseek),
 and that is called out where it matters.
 
+## Exactly-once (Kafka EOS)
+
+Most ingress libraries stop at offset EOS: the broker commits offsets in a
+transaction so a record is consumed once. `source` goes one step further on
+Kafka and ties the records a transition emits to that same commit, so a
+consume-process-produce cycle is atomic: the produced records and the consumed
+offset are committed together, or neither is.
+
+This is the `Transactional` capability, satisfied by the Kafka adapter only.
+JetStream and the in-memory adapter do not implement it, so the capability is
+absent rather than faked; on those backends use the at-least-once path. Build a
+transactional inlet with a stable, unique transactional id:
+
+```go
+in, _ := kafka.New(
+    kafka.WithSeedBrokers("localhost:9092"),
+    kafka.WithTransactional("orders-eos-v1"), // the producer fencing token
+)
+sub, _ := in.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}, Group: "orders-svc"})
+
+if tx, ok := sub.(source.Transactional); ok {
+    // Begin fences the produce side and the offset of m in one transaction.
+    err := tx.Begin(ctx, m, func(ctx context.Context, t source.Tx) error {
+        return t.Produce(ctx, source.ProducedRecord{Topic: "orders.out", Value: out})
+    })
+    // err == nil means the produced records AND m's offset committed atomically.
+}
+```
+
+`Begin` runs your work function inside a Kafka producer transaction. If it
+returns nil, the records you produced through the `Tx` are flushed and the
+consumed offset is committed as one unit. If it returns an error, or a rebalance
+fences the producer, the transaction aborts: the produced records are discarded,
+the offset is not committed, and the input is redelivered. `Begin` is the full
+settle path for that message, so the engine takes no further settle action.
+
+Under the hood the adapter builds a franz-go `GroupTransactSession` with
+read-committed fetch isolation (so it never reads another producer's uncommitted
+records) and no auto-commit (the transaction commits offsets). No franz-go type
+appears in the seam: you produce neutral `source.ProducedRecord` values, and the
+adapter maps them to the wire.
+
+### Limits
+
+- **Kafka only.** EOS here is the Kafka transactional protocol. JetStream and
+  other backends fall back to the at-least-once path; they do not pretend to
+  offer it.
+- **One transactional id per logical consumer.** The id is a producer fencing
+  token. Two live consumers sharing one id fence each other; keep it stable
+  across restarts and unique per consumer.
+- **Idempotent downstream still matters across the persist gap.** When the bridge
+  persists machine state inside the transaction, a broker abort after a
+  successful save can leave the instance advanced but the offset uncommitted. The
+  redelivery is then deduplicated by the machine's state version (see
+  [driving a statechart with EOS](/crucible/source/with-state/#exactly-once-consume-process-produce-kafka-eos)),
+  so it acks as a no-op rather than double-applying.
+
 ## Roadmap and bring your own
 
 `source/redis` (Redis Streams) and `source/cdc` (Debezium/OpenCDC

docs/src/content/docs/source/with-state.md

@@ -65,9 +65,48 @@ here it is a typed outcome.
 ## Consume, transition, emit
 
 A transition's emitted effects can be handed to a
-[`sink`](/crucible/sink/overview/) Manifold in the same step, optionally
-transactional on Kafka EOS. The statechart is the processor; the consume loop and
-the fan-out are the two ends of it.
+[`sink`](/crucible/sink/overview/) Manifold in the same step. The statechart is
+the processor; the consume loop and the fan-out are the two ends of it. The
+default bridge (`Drive`) hands effects to a `Sink` and acks after a durable save,
+which is at-least-once: an effect can be emitted twice if the process dies after
+emit but before the offset commits.
+
+## Exactly-once consume-process-produce (Kafka EOS)
+
+On a `Transactional` subscription (Kafka built with `kafka.WithTransactional`),
+`DriveTx` closes that gap. It fires the transition, produces the emitted effects,
+persists the new state, and commits the consumed offset all inside one Kafka
+transaction, so the emitted records and the ack of the message that produced them
+are exactly-once. Effects are produced through a `TxSink`, the transactional
+mirror of `Sink`, which turns each effect into records on the open transaction:
+
+```go
+txSink := statemachine.TxSinkFunc(func(ctx context.Context, tx source.Tx, eff any) error {
+    if oe, ok := eff.(OpenedEffect); ok {
+        return tx.Produce(ctx, source.ProducedRecord{Topic: "turnstile.out", Value: encode(oe)})
+    }
+    return fmt.Errorf("unmapped effect %T", eff) // fail loudly, never drop an output
+})
+
+h := statemachine.DriveTx(machine, store, router, transactionalSub, txSink)
+// On success h returns source.Manual: the transaction already committed the offset.
+```
+
+On a successful transition `DriveTx` returns `source.Manual`, because the
+transaction committed the offset itself and the engine must not settle again. A
+route failure is still `Term`, a guard rejection still `Reject`, and a redelivered
+event already folded into the version is still a `Skip` (a plain offset advance,
+not a transaction). An emit failure, a persist failure, or a broker abort (a
+rebalance fences the producer) is a `Nak`: nothing committed, and the input is
+redelivered.
+
+One subtlety is honest rather than hidden: the durable save runs inside the
+transaction, so a broker abort after a successful save can leave the instance
+advanced but the offset uncommitted. The redelivery is then deduplicated by the
+machine's state version (the same exactly-once-into-the-machine mechanism above),
+so it acks as a no-op rather than double-applying. EOS is Kafka only; on
+JetStream and other backends the `Transactional` capability is absent and `Drive`
+gives the at-least-once path.
 
 ## Analyzable consumption
 

source/capability.go

@@ -113,14 +113,66 @@ type Batched interface {
 	SettleBatch(ctx context.Context, ms []Message, r Result) error
 }
 
-// Transactional is a [Subscription] that can fence message settlement inside a
-// transaction, so consume-process-produce is atomic (exactly-once into a sink).
+// Transactional is a [Subscription] that can fence message settlement and the
+// records produced while processing inside one transaction, so a
+// consume-process-produce cycle is atomic: the produced records and the consumed
+// offsets are committed together or not at all (exactly-once into a sink).
 // Satisfied by Kafka (EOS) only. JetStream has no equivalent and does NOT
 // implement it — the capability is absent rather than faked.
+//
+// The contract is read-only on the message side: a transaction does not change
+// how a delivered [Message] is settled (the engine still marks/commits the
+// consumed offsets); it adds an atomic boundary that ties any records produced
+// through the [Tx] to that settlement. Use it through the state-machine bridge
+// (Drive) or directly: assert a subscription to Transactional, then run the
+// produce side of the work inside [Transactional.Begin].
 type Transactional interface {
-	// Begin starts a transaction; settlement of messages received during it is
-	// committed or aborted atomically with the work done inside fn.
-	Begin(ctx context.Context, fn func(ctx context.Context) error) error
+	// Begin starts a transaction around processing the consumed message m and runs
+	// fn inside it, handing fn a [Tx] to produce records on. If fn returns nil the
+	// transaction commits: every record produced through tx is flushed AND m's
+	// consumed offset is committed in one atomic unit, so the emitted records and
+	// the ack of the input that produced them are exactly-once. If fn returns an
+	// error, or the consumer is fenced by a rebalance, the transaction aborts: the
+	// produced records are discarded and m's offset is not committed, so m is
+	// redelivered. Begin returns fn's error on abort, or a non-nil error if the
+	// commit itself failed.
+	//
+	// m must be a message this subscription delivered. Begin is the only settle
+	// path for m when used transactionally: the caller does NOT also call
+	// [Subscription.Settle] for m, because Begin commits its offset itself.
+	Begin(ctx context.Context, m Message, fn func(ctx context.Context, tx Tx) error) error
+}
+
+// Tx is the backend-neutral produce handle a [Transactional.Begin] hands to its
+// work function: records produced through it are buffered into the in-flight
+// transaction and committed atomically with the consumed offsets, or discarded
+// if the transaction aborts. It carries no vendor type; a power user who needs
+// the native producer reaches it through the subscription's As escape hatch
+// instead. A Tx is valid only for the duration of the Begin call that produced
+// it; using it afterward is a programming error.
+type Tx interface {
+	// Produce enqueues records into the open transaction. They are not visible to
+	// a read-committed consumer until the transaction commits. A non-nil error
+	// means the produce failed (so the work function should return it, aborting the
+	// transaction); a nil error means the records are buffered, not yet committed.
+	Produce(ctx context.Context, records ...ProducedRecord) error
+}
+
+// ProducedRecord is a backend-neutral record emitted inside a [Tx]: a topic (or
+// subject), an optional partitioning key, a value, and optional headers. It is
+// the produce-side mirror of the read-only [Message] the consumer delivers, kept
+// free of vendor types so the bridge can build it without importing an adapter.
+type ProducedRecord struct {
+	// Topic is the destination topic (Kafka) or subject the record is produced to.
+	// It is required; an empty Topic is a programming error.
+	Topic string
+	// Key is the optional partitioning key. An empty key lets the backend choose a
+	// partition.
+	Key []byte
+	// Value is the record payload.
+	Value []byte
+	// Headers are optional record headers carried alongside the value.
+	Headers Headers
 }
 
 // Deduper is a [Subscription] (or an inlet seam) that suppresses re-delivery of

source/capability_test.go

@@ -30,11 +30,17 @@ func (fullCap) SettleBatch(context.Context, []source.Message, source.Result) err
 	return nil
 }
 
-func (fullCap) Begin(_ context.Context, fn func(context.Context) error) error {
-	return fn(context.Background())
+func (fullCap) Begin(_ context.Context, _ source.Message, fn func(context.Context, source.Tx) error) error {
+	return fn(context.Background(), txStub{})
 }
-func (fullCap) Seen(context.Context, string) (bool, error) { return false, nil }
-func (fullCap) Lag(context.Context) (int64, error)         { return 0, nil }
+
+// txStub is a no-op [source.Tx] that records nothing; it satisfies the produce
+// handle the capability hands to a transactional work function.
+type txStub struct{}
+
+func (txStub) Produce(context.Context, ...source.ProducedRecord) error { return nil }
+func (fullCap) Seen(context.Context, string) (bool, error)             { return false, nil }
+func (fullCap) Lag(context.Context) (int64, error)                     { return 0, nil }
 
 func TestCapabilities_DetectedByTypeAssertion(t *testing.T) {
 	t.Parallel()
@@ -72,7 +78,7 @@ func TestCapabilities_DetectedByTypeAssertion(t *testing.T) {
 	_ = c.SeekToEnd(context.Background())
 	_, _ = c.NextBatch(context.Background(), 1)
 	_ = c.SettleBatch(context.Background(), nil, source.Ack())
-	_ = c.Begin(context.Background(), func(context.Context) error { return nil })
+	_ = c.Begin(context.Background(), nil, func(context.Context, source.Tx) error { return nil })
 	_, _ = c.Seen(context.Background(), "k")
 	_, _ = c.Lag(context.Background())
 }

source/kafka/CHANGELOG.md

@@ -25,5 +25,13 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.
   `Transactional` (Kafka EOS). `BlockRebalanceOnPoll` provides a safe
   processing window. The underlying `*kgo.Client` and `*kgo.Record` are
   reachable only through `As`.
+- Exactly-once consume-process-produce. `WithTransactional(id)` builds a
+  `GroupTransactSession` with read-committed fetch isolation and no auto-commit,
+  and the subscription's `source.Transactional.Begin(ctx, m, fn)` runs `fn`
+  inside a producer transaction: records produced through the handed `source.Tx`
+  are flushed and `m`'s consumed offset is committed in one atomic unit, or the
+  transaction aborts (on a work error or a rebalance fence) and the input is
+  redelivered. Produced records are neutral `source.ProducedRecord` values; no
+  franz-go type crosses the seam.
 
 [Unreleased]: https://github.com/stablekernel/crucible/tree/main/source/kafka