From ecc11cf765dbd1ec1e59d5706b4196ee0d35abd1 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Wed, 3 Jun 2026 19:23:48 -0400 Subject: [PATCH] feat: add Kafka exactly-once consume-process-produce Flesh out the source Transactional capability into a working EOS path so a consume-process-produce cycle commits the produced records and the consumed offset atomically on Kafka. Core seam (source): Transactional.Begin now fences a consumed Message together with the records produced through a new backend-neutral Tx handle (Tx.Produce, ProducedRecord), so emit and ack are one atomic unit. No vendor type crosses the seam. source/kafka: WithTransactional(id) builds a GroupTransactSession with read-committed isolation and no auto-commit; Begin produces through the session, marks the consumed offset on success only, and ends with TryCommit/TryAbort. A narrow transactor seam keeps the begin/produce/end choreography unit-testable with a fake session. source/statemachine: DriveTx runs Fire + emit-as-produce + persist + offset commit inside one transaction via a TxSink, returning Manual so the engine does not re-settle; Drive remains the at-least-once fallback. A broker abort after a durable save is covered by state-version dedup. Tests: fake transact-session unit tests asserting begin/produce/commit/abort ordering and the bridge's transactional path; an integration test (RedPanda) doing a consume-produce EOS round-trip and asserting no duplicate output across an aborted transaction. Docs: an Exactly-once (Kafka EOS) section in source adapters and with-state, plus changelog notes. Signed-off-by: Joshua Temple --- docs/src/content/docs/source/adapters.md | 62 ++++- docs/src/content/docs/source/with-state.md | 45 +++- source/capability.go | 62 ++++- source/capability_test.go | 16 +- source/kafka/CHANGELOG.md | 8 + source/kafka/adapter_internal_test.go | 243 +++++++++++++++++++- source/kafka/capability.go | 109 ++++++++- source/kafka/integration_test.go | 148 ++++++++++++ source/kafka/kafka.go | 57 ++++- source/kafka/subscription.go | 8 +- source/statemachine/CHANGELOG.md | 7 + source/statemachine/drivetx.go | 186 +++++++++++++++ source/statemachine/drivetx_test.go | 255 +++++++++++++++++++++ source/statemachine/example_test.go | 61 +++++ 14 files changed, 1234 insertions(+), 33 deletions(-) create mode 100644 source/statemachine/drivetx.go create mode 100644 source/statemachine/drivetx_test.go diff --git a/docs/src/content/docs/source/adapters.md b/docs/src/content/docs/source/adapters.md index ead8748..6522fe3 100644 --- a/docs/src/content/docs/source/adapters.md +++ b/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 diff --git a/docs/src/content/docs/source/with-state.md b/docs/src/content/docs/source/with-state.md index 86c18e8..705d9de 100644 --- a/docs/src/content/docs/source/with-state.md +++ b/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 diff --git a/source/capability.go b/source/capability.go index 9ccca1e..6243ad3 100644 --- a/source/capability.go +++ b/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 diff --git a/source/capability_test.go b/source/capability_test.go index 924dab9..20092a5 100644 --- a/source/capability_test.go +++ b/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()) } diff --git a/source/kafka/CHANGELOG.md b/source/kafka/CHANGELOG.md index 25e2fab..29b1978 100644 --- a/source/kafka/CHANGELOG.md +++ b/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 diff --git a/source/kafka/adapter_internal_test.go b/source/kafka/adapter_internal_test.go index b378c86..b4fa130 100644 --- a/source/kafka/adapter_internal_test.go +++ b/source/kafka/adapter_internal_test.go @@ -6,6 +6,7 @@ import ( "context" "crypto/tls" "errors" + "sync" "testing" "time" @@ -59,16 +60,26 @@ func TestOptionsApply(t *testing.T) { WithDLQTopic("orders.DLQ")(&cfg) WithMaxPollRecords(128)(&cfg) WithBalancer(kgo.CooperativeStickyBalancer())(&cfg) - WithTransactional()(&cfg) + WithTransactional("orders-eos-v1")(&cfg) if len(cfg.seedBrokers) != 2 || cfg.clientID != "crucible" || cfg.dlqTopic != "orders.DLQ" { t.Errorf("config = %+v, want seeds/clientID/dlq applied", cfg) } - if cfg.maxPoll != 128 || len(cfg.balancers) != 1 || !cfg.transact { + if cfg.maxPoll != 128 || len(cfg.balancers) != 1 || !cfg.transact || cfg.transactID != "orders-eos-v1" { t.Errorf("config = %+v, want maxPoll/balancer/transact applied", cfg) } } +func TestWithTransactionalEmptyIDIgnored(t *testing.T) { + t.Parallel() + + cfg := config{} + WithTransactional("")(&cfg) + if cfg.transact || cfg.transactID != "" { + t.Errorf("WithTransactional(\"\") = %+v, want non-transactional", cfg) + } +} + func TestSubscribeRejectsNoTopics(t *testing.T) { t.Parallel() @@ -365,12 +376,238 @@ func TestBeginWithoutTransactionalFails(t *testing.T) { t.Parallel() sub := newSub(&fakePoller{}) - err := sub.Begin(context.Background(), func(context.Context) error { return nil }) + m := newMessage(rec("orders", 0, 0, "k", "v")) + err := sub.Begin(context.Background(), m, func(context.Context, source.Tx) error { return nil }) if !errors.Is(err, errNotTransactional) { t.Fatalf("Begin() = %v, want errNotTransactional", err) } } +// fakeTransactor is a hand-rolled [transactor] that records the begin/produce/end +// call sequence so the EOS choreography is asserted with no broker. It returns a +// scripted commit/abort outcome and optional errors at each step. +type fakeTransactor struct { + mu sync.Mutex + + calls []string // ordered record of begin/produce/end + produced [][]*kgo.Record // records per ProduceSync call + beginErr error + produceTr error + endErr error + committed bool +} + +func (f *fakeTransactor) Begin() error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "begin") + return f.beginErr +} + +func (f *fakeTransactor) ProduceSync(_ context.Context, rs ...*kgo.Record) kgo.ProduceResults { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "produce") + f.produced = append(f.produced, rs) + out := make(kgo.ProduceResults, len(rs)) + for i, r := range rs { + out[i] = kgo.ProduceResult{Record: r, Err: f.produceTr} + } + return out +} + +func (f *fakeTransactor) End(_ context.Context, commit kgo.TransactionEndTry) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if commit == kgo.TryCommit { + f.calls = append(f.calls, "end-commit") + } else { + f.calls = append(f.calls, "end-abort") + } + return f.committed, f.endErr +} + +func newTxSub(ft *fakeTransactor) (*subscription, *fakePoller) { + fp := &fakePoller{} + return &subscription{client: fp, group: "g", dlqTopic: "orders.DLQ", transactSess: ft}, fp +} + +func TestBeginCommitsOnSuccessAfterProduce(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, fp := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + return tx.Produce(ctx, source.ProducedRecord{ + Topic: "orders.out", + Key: []byte("A-1"), + Value: []byte("emitted"), + Headers: source.Headers{{Key: "message-id", Value: "evt-1"}}, + }) + }) + if err != nil { + t.Fatalf("Begin() error = %v, want nil", err) + } + // The consumed record is marked (inside the tx, before End-commit) so its + // offset commits atomically with the produced record. + if got := fp.markedCount(); got != 1 { + t.Errorf("marked = %d, want 1 (consumed offset marked on commit)", got) + } + if got, want := ft.calls, []string{"begin", "produce", "end-commit"}; !equalStrings(got, want) { + t.Fatalf("call order = %v, want %v", got, want) + } + if len(ft.produced) != 1 || len(ft.produced[0]) != 1 { + t.Fatalf("produced = %#v, want one record in one ProduceSync", ft.produced) + } + pr := ft.produced[0][0] + if pr.Topic != "orders.out" || string(pr.Key) != "A-1" || string(pr.Value) != "emitted" { + t.Errorf("produced record = %+v, want orders.out/A-1/emitted", pr) + } + if len(pr.Headers) != 1 || pr.Headers[0].Key != "message-id" || string(pr.Headers[0].Value) != "evt-1" { + t.Errorf("produced headers = %+v, want message-id=evt-1", pr.Headers) + } +} + +func TestBeginAbortsAndReturnsWorkError(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{} + sub, fp := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + boom := errors.New("handler boom") + + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + _ = tx.Produce(ctx, source.ProducedRecord{Topic: "orders.out", Value: []byte("v")}) + return boom + }) + if !errors.Is(err, boom) { + t.Fatalf("Begin() = %v, want the work error", err) + } + // On abort the consumed record is NOT marked, so the input is redelivered. + if got := fp.markedCount(); got != 0 { + t.Errorf("marked = %d on abort, want 0 (offset not committed)", got) + } + if got, want := ft.calls, []string{"begin", "produce", "end-abort"}; !equalStrings(got, want) { + t.Fatalf("call order = %v, want %v (abort on work error)", got, want) + } +} + +func TestBeginAbortsWhenProduceFails(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{produceTr: errors.New("produce rejected")} + sub, fp := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + // A failed produce is returned by the work function, aborting the tx. + return tx.Produce(ctx, source.ProducedRecord{Topic: "orders.out", Value: []byte("v")}) + }) + if err == nil { + t.Fatal("Begin() = nil, want the produce error") + } + if got := fp.markedCount(); got != 0 { + t.Errorf("marked = %d on produce failure, want 0", got) + } + if got, want := ft.calls, []string{"begin", "produce", "end-abort"}; !equalStrings(got, want) { + t.Fatalf("call order = %v, want %v", got, want) + } +} + +func TestBeginReportsBrokerAbortOnSuccessfulWork(t *testing.T) { + t.Parallel() + + // Work succeeds but End reports the transaction did not commit (a fence/abort + // by the broker, e.g. a rebalance): Begin surfaces errTransactionAborted so + // the caller treats the cycle as not-committed and the input is redelivered. + ft := &fakeTransactor{committed: false} + sub, _ := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + + err := sub.Begin(context.Background(), m, func(context.Context, source.Tx) error { return nil }) + if !errors.Is(err, errTransactionAborted) { + t.Fatalf("Begin() = %v, want errTransactionAborted", err) + } + if got, want := ft.calls, []string{"begin", "end-commit"}; !equalStrings(got, want) { + t.Fatalf("call order = %v, want %v", got, want) + } +} + +func TestBeginPropagatesBeginError(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{beginErr: errors.New("cannot begin")} + sub, _ := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + ran := false + err := sub.Begin(context.Background(), m, func(context.Context, source.Tx) error { + ran = true + return nil + }) + if err == nil { + t.Fatal("Begin() = nil, want the begin error") + } + if ran { + t.Error("work function ran despite a begin failure") + } +} + +func TestBeginRejectsForeignMessage(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, _ := newTxSub(ft) + err := sub.Begin(context.Background(), foreignMessage{}, func(context.Context, source.Tx) error { return nil }) + if !errors.Is(err, errNotKafkaMessage) { + t.Fatalf("Begin(foreign) = %v, want errNotKafkaMessage", err) + } +} + +func TestTxProduceRejectsEmptyTopic(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, _ := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + return tx.Produce(ctx, source.ProducedRecord{Value: []byte("no topic")}) + }) + if !errors.Is(err, errEmptyTopic) { + t.Fatalf("Begin() = %v, want errEmptyTopic", err) + } +} + +func TestTxProduceNoRecordsIsNoOp(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, _ := newTxSub(ft) + m := newMessage(rec("orders", 0, 7, "A-1", "placed")) + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + return tx.Produce(ctx) // no records + }) + if err != nil { + t.Fatalf("Begin() error = %v, want nil", err) + } + if got, want := ft.calls, []string{"begin", "end-commit"}; !equalStrings(got, want) { + t.Fatalf("call order = %v, want %v (no produce for an empty batch)", got, want) + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + func TestPartitionOrderedMarker(t *testing.T) { t.Parallel() // The marker method is the guarantee; calling it must be a harmless no-op. diff --git a/source/kafka/capability.go b/source/kafka/capability.go index dc7b210..1b1b2e8 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -30,6 +30,9 @@ var ( // errTransactionAborted reports that End returned without committing despite // the work function succeeding (a broker-side abort). errTransactionAborted = errors.New("source/kafka: transaction aborted by broker") + // errEmptyTopic reports that a ProducedRecord with no Topic was produced into + // a transaction; the destination is required. + errEmptyTopic = errors.New("source/kafka: produced record has no topic") ) // requester is the franz-go surface the seek/lag capabilities need beyond the @@ -251,19 +254,111 @@ func (s *subscription) Lag(ctx context.Context) (int64, error) { // --- Transactional ---------------------------------------------------------- -// Begin runs fn inside a Kafka producer transaction so consumed records settled -// during it are committed (or aborted) atomically with the produces fn performs -// — exactly-once consume-process-produce. It is available only when the inlet -// was built with [WithTransactional]; otherwise it reports the capability is -// absent rather than silently running fn without a transaction. -func (s *subscription) Begin(ctx context.Context, fn func(ctx context.Context) error) error { +// transactor is the narrow franz-go transaction surface the EOS path drives: a +// *kgo.GroupTransactSession satisfies it. Narrowing to it keeps the begin → +// produce → end choreography unit-testable with a fake session and no broker. +// ProduceSync produces records into the open transaction; End commits or aborts, +// committing the consumed offsets atomically with the produced records when it +// commits. +type transactor interface { + Begin() error + ProduceSync(ctx context.Context, rs ...*kgo.Record) kgo.ProduceResults + End(ctx context.Context, commit kgo.TransactionEndTry) (committed bool, err error) +} + +// Compile-time proof the real session satisfies the narrow seam. +var _ transactor = (*kgo.GroupTransactSession)(nil) + +// txHandle is the [source.Tx] the EOS path hands to a transactional work +// function: it produces records into the open transaction through the +// [transactor]. It buffers nothing of its own — franz-go holds the in-flight +// records until End commits or aborts them — so Produce is a direct, synchronous +// produce into the transaction whose error the work function propagates to +// trigger an abort. A txHandle is valid only for the Begin call that created it. +type txHandle struct { + sess transactor +} + +// Produce emits records into the open transaction. The neutral +// [source.ProducedRecord]s are mapped onto franz-go records and produced +// synchronously; a produce error is returned so the work function can abort the +// transaction by returning it. Produced records are not visible to a +// read-committed consumer until the transaction commits. +func (t txHandle) Produce(ctx context.Context, records ...source.ProducedRecord) error { + if len(records) == 0 { + return nil + } + recs := make([]*kgo.Record, 0, len(records)) + for _, pr := range records { + if pr.Topic == "" { + return fmt.Errorf("source/kafka: produce in transaction: %w", errEmptyTopic) + } + recs = append(recs, &kgo.Record{ + Topic: pr.Topic, + Key: pr.Key, + Value: pr.Value, + Headers: toRecordHeaders(pr.Headers), + }) + } + if err := t.sess.ProduceSync(ctx, recs...).FirstErr(); err != nil { + return fmt.Errorf("source/kafka: produce in transaction: %w", err) + } + return nil +} + +// toRecordHeaders maps neutral [source.Headers] onto franz-go record headers. +func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { + if len(hs) == 0 { + return nil + } + out := make([]kgo.RecordHeader, len(hs)) + for i, h := range hs { + out[i] = kgo.RecordHeader{Key: h.Key, Value: []byte(h.Value)} + } + return out +} + +// Begin runs fn inside a Kafka producer transaction so the records fn produces +// through the handed [source.Tx] are committed (or aborted) atomically with the +// consumed offset of m — exactly-once consume-process-produce. The choreography +// is: begin the transaction, run fn (which produces emitted records through the +// txHandle), mark m's offset for commit on success, then End with TryCommit if fn +// succeeded or TryAbort if it failed. On commit, franz-go flushes the produced +// records and commits the group's marked offsets (including m's) in one unit; on +// abort it discards the produced records and does not commit m's offset, so m is +// redelivered. +// +// m's offset is marked only after fn succeeds and only inside the transaction, so +// a failed transform never advances the consumed position: the commit-after- +// process invariant holds, now atomically with the produce side. Begin is the +// full settle path for m; the caller must not also call [Subscription.Settle] for +// m. +// +// It is available only when the inlet was built with [WithTransactional]; +// otherwise it reports the capability is absent rather than silently running fn +// 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. +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) } + rec, ok := recordOf(m) + if !ok { + return fmt.Errorf("source/kafka: transactional begin: %w", errNotKafkaMessage) + } + defer s.settled() + if err := s.transactSess.Begin(); err != nil { return fmt.Errorf("source/kafka: begin transaction: %w", err) } - fnErr := fn(ctx) + fnErr := fn(ctx, txHandle{sess: s.transactSess}) + if fnErr == nil { + // Mark the consumed record only on success and inside the open + // transaction, so End commits the produced records and this offset as one + // unit. A failed transform leaves the offset unmarked, so m is redelivered. + s.client.MarkCommitRecords(rec) + } commit := kgo.TryCommit if fnErr != nil { commit = kgo.TryAbort diff --git a/source/kafka/integration_test.go b/source/kafka/integration_test.go index 1302956..a3cd541 100644 --- a/source/kafka/integration_test.go +++ b/source/kafka/integration_test.go @@ -143,6 +143,154 @@ func TestIntegrationConsumeAckTermRoundTrip(t *testing.T) { } } +// TestIntegrationTransactionalEOSRoundTrip starts a real RedPanda broker and +// proves exactly-once consume-process-produce: it produces an input record, +// consumes it through a transactional Inlet, and runs two transactions for it. +// The first transaction produces an output record but is aborted (its work +// function fails), so neither the output record nor the input offset is +// committed; a read-committed reader sees no output, and the input is +// redelivered. The second transaction produces the same output and commits, so +// exactly one output record exists and the offset advances. The aborted attempt +// leaves no duplicate. +func TestIntegrationTransactionalEOSRoundTrip(t *testing.T) { + skipWithoutDocker(t) + + ctx := context.Background() + container, err := redpanda.Run(ctx, redpandaImage) + if err != nil { + t.Skipf("redpanda.Run unavailable (image pull or startup failed); skipping: %v", err) + } + t.Cleanup(func() { _ = testcontainers.TerminateContainer(container) }) + + broker, err := container.KafkaSeedBroker(ctx) + if err != nil { + t.Fatalf("KafkaSeedBroker() error = %v", err) + } + + const ( + inTopic = "eos.in" + outTopic = "eos.out" + group = "eos-consumer" + txID = "eos-it-v1" + ) + createTopics(ctx, t, broker, inTopic, outTopic) + + prod, err := kgo.NewClient(kgo.SeedBrokers(broker)) + if err != nil { + t.Fatalf("producer client error = %v", err) + } + t.Cleanup(prod.Close) + produce(ctx, t, prod, inTopic, "K-1", "input") + + inlet, err := kafkasource.New( + kafkasource.WithSeedBrokers(broker), + kafkasource.WithClientID("eos-consumer"), + kafkasource.WithTransactional(txID), + kafkasource.WithClientOptions(kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + t.Cleanup(func() { _ = inlet.Close() }) + + sub, err := inlet.Subscribe(ctx, source.SubscribeConfig{Topics: []string{inTopic}, Group: group}) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + tx, ok := sub.(source.Transactional) + if !ok { + t.Fatal("subscription does not satisfy source.Transactional with WithTransactional") + } + + pollCtx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + + // First receive: open a transaction that produces the output, then fail the + // work function so the transaction aborts. Nothing should be committed. + m1, err := sub.Next(pollCtx) + if err != nil { + t.Fatalf("Next() #1 error = %v", err) + } + if string(m1.Key()) != "K-1" { + t.Fatalf("first record key = %q, want K-1", m1.Key()) + } + abortErr := errors.New("deliberate abort") + beginErr := tx.Begin(pollCtx, m1, func(c context.Context, txn source.Tx) error { + if perr := txn.Produce(c, source.ProducedRecord{Topic: outTopic, Key: []byte("K-1"), Value: []byte("output")}); perr != nil { + return perr + } + return abortErr // abort: discard the produced record, do not commit the offset + }) + if !errors.Is(beginErr, abortErr) { + t.Fatalf("Begin() #1 = %v, want the abort error", beginErr) + } + + // The input must be redelivered because its offset was not committed. Re-seek + // the subscription to start so the redelivery is deterministic even though the + // aborted transaction left the consumer position past the record. + if sk, ok := sub.(source.Seekable); ok { + if serr := sk.SeekToStart(pollCtx); serr != nil { + t.Fatalf("SeekToStart() error = %v", serr) + } + } + + m2, err := sub.Next(pollCtx) + if err != nil { + t.Fatalf("Next() #2 (redelivery) error = %v", err) + } + if string(m2.Key()) != "K-1" { + t.Fatalf("redelivered key = %q, want K-1 (aborted offset not committed)", m2.Key()) + } + + // Second transaction: produce the same output and commit. This is the only + // transaction that lands an output record and advances the offset. + commitErr := tx.Begin(pollCtx, m2, func(c context.Context, txn source.Tx) error { + return txn.Produce(c, source.ProducedRecord{Topic: outTopic, Key: []byte("K-1"), Value: []byte("output")}) + }) + if commitErr != nil { + t.Fatalf("Begin() #2 (commit) error = %v", commitErr) + } + + if cerr := sub.Close(); cerr != nil { + t.Fatalf("Close() error = %v", cerr) + } + + // A read-committed reader must see exactly one output record: the aborted + // transaction's produce is invisible, so there is no duplicate. + outClient, err := kgo.NewClient( + kgo.SeedBrokers(broker), + kgo.ConsumeTopics(outTopic), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + kgo.FetchIsolationLevel(kgo.ReadCommitted()), + ) + if err != nil { + t.Fatalf("out client error = %v", err) + } + t.Cleanup(outClient.Close) + + var got []*kgo.Record + readCtx, readCancel := context.WithTimeout(ctx, 20*time.Second) + defer readCancel() + for len(got) < 1 { + f := outClient.PollFetches(readCtx) + if errs := f.Errors(); len(errs) > 0 { + for _, fe := range errs { + if fe.Err != nil && !errors.Is(fe.Err, context.DeadlineExceeded) { + t.Fatalf("out PollFetches error = %v", fe.Err) + } + } + break + } + got = append(got, f.Records()...) + } + if len(got) != 1 { + t.Fatalf("committed output records = %d, want exactly 1 (no duplicate across the aborted transaction)", len(got)) + } + if string(got[0].Key) != "K-1" || string(got[0].Value) != "output" { + t.Errorf("output record = %s/%s, want K-1/output", got[0].Key, got[0].Value) + } +} + // createTopics creates the given topics (one partition, replication factor one) // against the broker and waits for the admin call to succeed, so produces never // race auto-creation. An already-exists result is treated as success. diff --git a/source/kafka/kafka.go b/source/kafka/kafka.go index 4ee69b1..04bfcda 100644 --- a/source/kafka/kafka.go +++ b/source/kafka/kafka.go @@ -81,6 +81,7 @@ type config struct { balancers []kgo.GroupBalancer maxPoll int transact bool + transactID string extraOpts []kgo.Opt client *kgo.Client } @@ -177,12 +178,25 @@ func WithMaxPollRecords(n int) Option { } // WithTransactional enables Kafka exactly-once semantics: the subscription -// satisfies [source.Transactional], fencing settlement of consumed records -// inside a producer transaction so consume-process-produce is atomic. It -// requires a transactional ID, which franz-go derives; the inlet builds a -// kgo.GroupTransactSession instead of a plain client. -func WithTransactional() Option { - return func(c *config) { c.transact = true } +// satisfies [source.Transactional], fencing the records produced while +// processing and the consumed offset inside one producer transaction so +// consume-process-produce is atomic. The inlet builds a kgo.GroupTransactSession +// instead of a plain client, configured with the given transactional ID, +// read-committed fetch isolation (so it never reads another producer's +// uncommitted records), and no auto-commit (the transaction commits offsets). +// +// transactionalID must be stable and unique to this logical consumer across +// restarts: it is the producer fencing token Kafka uses to detect a zombie +// instance. Reusing one transactional ID across two live consumers of the same +// partitions fences one of them. An empty transactionalID is ignored and the +// inlet stays non-transactional. +func WithTransactional(transactionalID string) Option { + return func(c *config) { + if transactionalID != "" { + c.transact = true + c.transactID = transactionalID + } + } } // WithClientOptions appends raw franz-go options for tuning this package does @@ -290,6 +304,32 @@ func (in *Inlet) consumeOpts(sc source.SubscribeConfig) []kgo.Opt { return opts } +// transactOpts assembles the franz-go options for an EOS group transact session: +// the base group options without auto-commit (the transaction commits offsets), +// the transactional ID that fences a zombie producer, and read-committed fetch +// isolation so the consumer never reads another producer's uncommitted records. +// Auto-committing is disabled during transactional consuming, so this path does +// NOT add AutoCommitMarks; the engine still marks records, and End commits the +// marked offsets transactionally. +func (in *Inlet) transactOpts(sc source.SubscribeConfig) []kgo.Opt { + opts := in.baseOpts() + opts = append(opts, + kgo.ConsumeTopics(sc.Topics...), + kgo.TransactionalID(in.cfg.transactID), + kgo.FetchIsolationLevel(kgo.ReadCommitted()), + // A safe processing window: a rebalance cannot move partitions mid-batch, + // and franz-go aborts the in-flight transaction on a rebalance. + kgo.BlockRebalanceOnPoll(), + ) + if sc.Group != "" { + opts = append(opts, kgo.ConsumerGroup(sc.Group)) + } + if len(in.cfg.balancers) > 0 { + opts = append(opts, kgo.Balancers(in.cfg.balancers...)) + } + return opts +} + // Subscribe builds (or reuses) the franz-go client for cfg and returns a live // [source.Subscription]. At least one topic is required. The returned // subscription owns the rebalance hooks, so it must be installed before the @@ -306,6 +346,11 @@ func (in *Inlet) Subscribe(_ context.Context, cfg source.SubscribeConfig) (sourc if in.client == nil { opts := in.consumeOpts(cfg) + if in.cfg.transact { + // The transactional path uses read-committed isolation, a + // transactional ID, and no auto-commit; build its own option set. + opts = in.transactOpts(cfg) + } // The subscription's rebalance trampolines are installed at client // build time so assign/revoke callbacks registered later still fire. opts = append(opts, diff --git a/source/kafka/subscription.go b/source/kafka/subscription.go index 3301291..e3aa6f2 100644 --- a/source/kafka/subscription.go +++ b/source/kafka/subscription.go @@ -38,8 +38,12 @@ var _ poller = (*kgo.Client)(nil) // re-polling only once a buffer is exhausted, which is where // BlockRebalanceOnPoll's safe window opens and closes. type subscription struct { - client poller - transactSess *kgo.GroupTransactSession + client poller + // transactSess is the EOS transaction session, present only when the inlet + // was built with WithTransactional. It is narrowed to the transactor seam so + // the begin/produce/end choreography is testable with a fake; nil means the + // subscription is not transactional and Begin reports the capability absent. + transactSess transactor group string dlqTopic string maxPoll int diff --git a/source/statemachine/CHANGELOG.md b/source/statemachine/CHANGELOG.md index 31ec172..82f00a0 100644 --- a/source/statemachine/CHANGELOG.md +++ b/source/statemachine/CHANGELOG.md @@ -12,6 +12,13 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. - `Drive` binding: routes a `source.Message` to an instance key and event, loads the instance through a `Store`, fires the event, hands the emitted effects to a `Sink`, persists the new state, and acks only after the durable commit. +- `DriveTx` binding: the exactly-once consume-process-produce mode for a + `source.Transactional` subscription (Kafka EOS). It fires the transition, + produces the emitted effects through a `TxSink` onto the open transaction, + persists the new state, and lets the transaction commit the produced records + and the consumed offset atomically, returning `source.Manual`. A `TxSink` + (`TxSinkFunc`) maps each effect to records on a `source.Tx`. `ErrNotTransactional` + reports a nil `Transactional` or `TxSink` at wiring time. - `DriveFunc` binding: the stateless mode, firing each message against a caller-supplied `FireFunc` with no persistence. - `Store` interface and `Record` type for durable instance persistence, with the diff --git a/source/statemachine/drivetx.go b/source/statemachine/drivetx.go new file mode 100644 index 0000000..b7943dc --- /dev/null +++ b/source/statemachine/drivetx.go @@ -0,0 +1,186 @@ +// SPDX-License-Identifier: Apache-2.0 + +package statemachine + +import ( + "context" + "errors" + "fmt" + + "github.com/stablekernel/crucible/source" + "github.com/stablekernel/crucible/state" + "github.com/stablekernel/crucible/telemetry" +) + +// TxSink translates a transition's emitted effects into records produced on a +// transactional [source.Tx], so the emitted records and the consumed offset are +// committed exactly once. It is the transactional mirror of [Sink]: where Sink +// fires-and-forgets an effect at an external destination, a TxSink produces the +// effect into the open transaction the consume-process-produce cycle is fenced +// by, via [source.Tx.Produce]. +// +// EmitTx is called once per emitted effect, in order, inside the transaction. A +// non-nil error aborts the transaction (so the consumed offset is not committed +// and the input is redelivered) and is returned by the produce side of [DriveTx]. +// An effect a TxSink does not recognize should return an error rather than +// silently dropping it, so a missing mapping fails loudly rather than losing an +// output. +type TxSink interface { + // EmitTx produces the records that realize effect into tx. The effect's + // concrete type is a crucible/state effect value; a TxSink type-switches on it + // and calls tx.Produce with the resulting [source.ProducedRecord]s. Returning + // an error aborts the transaction. + EmitTx(ctx context.Context, tx source.Tx, effect any) error +} + +// TxSinkFunc adapts a plain function to a [TxSink]. +type TxSinkFunc func(ctx context.Context, tx source.Tx, effect any) error + +// EmitTx calls the underlying function. +func (f TxSinkFunc) EmitTx(ctx context.Context, tx source.Tx, effect any) error { + return f(ctx, tx, effect) +} + +// DriveTx binds a durable state-machine instance to a [source.Handler] with +// exactly-once consume-process-produce on a [source.Transactional] subscription +// (Kafka EOS): the records a transition emits and the consumed offset of the +// inbound message are committed in one atomic unit, so neither the emitted output +// nor the ack can survive without the other. +// +// It is the transactional counterpart of [Drive]. The route, load, fire, +// version-idempotency, and state-aware-rejection semantics are identical; the +// difference is the commit boundary. On a successful fire, DriveTx opens a +// transaction around the message with [source.Transactional.Begin] and, inside +// it, produces the emitted effects through the [TxSink], persists the new state +// through the [Store], and lets Begin commit the produced records together with +// the consumed offset. It then returns [source.Manual]: the [source.Hopper] takes +// no further settle action because Begin already committed the offset. +// +// Outcomes: +// +// - Route/decode failure → [source.Term] (poison), settled by the Hopper (no +// transaction is opened). +// - Redelivery already applied (message id equals the persisted LastEventID) → +// [source.Skip] (acked, never re-fired). The Hopper settles it; the skip is a +// plain offset advance, not a transaction. +// - Fire rejected as illegal for the current state → [source.Reject] (Term, +// InvalidForState), settled by the Hopper. +// - Begin/emit/persist failure, or a broker abort (a rebalance fences the +// producer) → [source.Nak] (Retryable): the transaction aborted, the offset +// was not committed, and the message is redelivered. +// - Success → [source.Manual] after Begin commits the emitted records and the +// consumed offset atomically. +// +// A note on persist ordering: the [Store.Save] runs inside the transaction, so a +// broker abort after a successful Save leaves the instance advanced but the +// offset uncommitted. The redelivery is then deduplicated by version +// (LastEventID), so it acks as a no-op rather than double-applying — the +// exactly-once-into-the-machine guarantee covers the gap the offset abort opens. +// +// Use DriveTx only with a subscription that satisfies [source.Transactional] +// (Kafka built with kafka.WithTransactional). On any other backend the capability +// is absent; use [Drive] for the at-least-once path. The Hopper must run the +// handler with [source.WithConcurrency](1) per key so one key's transactions do +// not interleave (its default per-key ordering already provides this). +func DriveTx[K comparable, E comparable, C any]( + machine *state.Machine[K, E, C], + store Store[K, K, E, C], + router Router[K, E], + tx source.Transactional, + sink TxSink, + opts ...Option, +) source.Handler { + cfg := newConfig(opts...) + if tx == nil || sink == nil { + // Fail loudly on a wiring error rather than panicking on the first message: + // every message naks with the configuration error so the misconfiguration + // surfaces in logs and metrics without losing data. + return func(context.Context, source.Message) source.Result { + return source.Nak(ErrNotTransactional) + } + } + return func(ctx context.Context, m source.Message) source.Result { + ctx, span := cfg.tracer.Start(ctx, cfg.spanName) + defer span.End() + + key, event, err := router(m) + if err != nil { + span.RecordError(err) + span.SetStatus(telemetry.StatusError, "route failed") + return source.Term(fmt.Errorf("statemachine: route message: %w", err)) + } + + rec, loaded, err := store.Load(ctx, key) + if err != nil { + span.RecordError(err) + span.SetStatus(telemetry.StatusError, "load failed") + return source.Nak(fmt.Errorf("statemachine: load instance: %w", err)) + } + + eventID := cfg.eventID(m) + if loaded && eventID != "" && rec.LastEventID == eventID { + // Already folded into the persisted version: ack as a no-op. This is a + // plain offset advance, not a transactional produce, so it is settled by + // the Hopper rather than through Begin. + span.SetStatus(telemetry.StatusOK, "") + return source.Skip() + } + + inst, err := instanceFor(machine, rec, loaded) + if err != nil { + span.RecordError(err) + span.SetStatus(telemetry.StatusError, "restore failed") + return source.Nak(fmt.Errorf("statemachine: restore instance: %w", err)) + } + + fromState := inst.Current() + res := inst.Fire(ctx, event) + if res.Err != nil { + if reject, ok := classifyFire(res.Err, event, fromState); ok { + span.RecordError(reject) + span.SetStatus(telemetry.StatusError, "invalid for state") + return source.Reject(reject) + } + span.RecordError(res.Err) + span.SetStatus(telemetry.StatusError, "fire failed") + return source.Nak(fmt.Errorf("statemachine: fire %v: %w", event, res.Err)) + } + + next := Record[K, E, C]{ + Snapshot: inst.Snapshot(), + Version: rec.Version + 1, + LastEventID: eventID, + } + + // One atomic unit: emit the effects as produced records, persist the new + // state, then let Begin commit the produced records and the consumed offset + // together. Any error inside fn aborts the transaction, so nothing is + // committed and the message is redelivered. + beginErr := tx.Begin(ctx, m, func(ctx context.Context, txn source.Tx) error { + for _, eff := range res.Effects { + if err := sink.EmitTx(ctx, txn, eff); err != nil { + return fmt.Errorf("statemachine: emit effect in transaction: %w", err) + } + } + if err := store.Save(ctx, key, next, rec.Version); err != nil { + return fmt.Errorf("statemachine: persist transition: %w", err) + } + return nil + }) + if beginErr != nil { + span.RecordError(beginErr) + span.SetStatus(telemetry.StatusError, "transaction failed") + return source.Nak(beginErr) + } + + span.SetStatus(telemetry.StatusOK, "") + // Begin committed the offset transactionally; the Hopper must not settle + // again. + return source.Manual() + } +} + +// ErrNotTransactional reports that [DriveTx] was given a [source.Transactional] +// that is nil, or a [TxSink] that is nil. It is a wiring error surfaced at the +// first message rather than a runtime data condition. Match it with errors.Is. +var ErrNotTransactional = errors.New("statemachine: DriveTx requires a non-nil Transactional and TxSink") diff --git a/source/statemachine/drivetx_test.go b/source/statemachine/drivetx_test.go new file mode 100644 index 0000000..b22c403 --- /dev/null +++ b/source/statemachine/drivetx_test.go @@ -0,0 +1,255 @@ +// SPDX-License-Identifier: Apache-2.0 + +package statemachine_test + +import ( + "context" + "errors" + "testing" + + "github.com/stablekernel/crucible/source" + "github.com/stablekernel/crucible/source/statemachine" +) + +// fakeTx is an in-memory [source.Tx] that records every produced record so a test +// can assert what the transactional emit produced. It also fails on demand to +// exercise the abort path. +type fakeTx struct { + produced []source.ProducedRecord + failOn string // topic to fail Produce on; "" never fails +} + +func (t *fakeTx) Produce(_ context.Context, records ...source.ProducedRecord) error { + for _, r := range records { + if t.failOn != "" && r.Topic == t.failOn { + return errors.New("produce rejected") + } + t.produced = append(t.produced, r) + } + return nil +} + +// fakeTransactional is an in-memory [source.Transactional] that drives the +// begin → fn → commit/abort choreography against a fakeTx, recording the call +// sequence and whether the work function committed. committed=false simulates a +// broker abort (a rebalance fence) despite a successful work function. +type fakeTransactional struct { + tx *fakeTx + committed bool // whether End reports a commit on a successful fn + calls []string +} + +func (s *fakeTransactional) Begin(ctx context.Context, _ source.Message, fn func(context.Context, source.Tx) error) error { + s.calls = append(s.calls, "begin") + err := fn(ctx, s.tx) + if err != nil { + s.calls = append(s.calls, "abort") + return err + } + s.calls = append(s.calls, "commit") + if !s.committed { + return errors.New("source: transaction aborted by broker") + } + return nil +} + +// emitOpenedToTopic is a TxSink that produces the openedEffect onto "turnstile.out". +func emitOpenedToTopic(_ context.Context, tx source.Tx, effect any) error { + if oe, ok := effect.(openedEffect); ok { + return tx.Produce(context.Background(), source.ProducedRecord{ + Topic: "turnstile.out", + Key: []byte("turnstile"), + Value: []byte("opened by " + oe.By), + Headers: source.Headers{{Key: "kind", Value: "opened"}}, + }) + } + return errors.New("unknown effect") +} + +func TestDriveTx_CommitsEmitAndOffsetAtomically(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + seedFunded(t, m, store) + + tx := &fakeTx{} + sub := &fakeTransactional{tx: tx, committed: true} + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + res := h(context.Background(), msg("evt-1", "c1")) + if res.Action != source.ActionManual { + t.Fatalf("action = %v, want manual (the transaction committed the offset; err=%v)", res.Action, res.Err) + } + // begin → (emit produced + persist) → commit, in order. + if got, want := sub.calls, []string{"begin", "commit"}; !equalCalls(got, want) { + t.Fatalf("call order = %v, want %v", got, want) + } + if len(tx.produced) != 1 { + t.Fatalf("produced = %#v, want one emitted record", tx.produced) + } + if tx.produced[0].Topic != "turnstile.out" || string(tx.produced[0].Value) != "opened by coin" { + t.Errorf("produced = %+v, want turnstile.out/'opened by coin'", tx.produced[0]) + } + // The transition was persisted inside the transaction. + rec, ok, _ := store.Load(context.Background(), keyOf) + if !ok || rec.Version != 2 || rec.Snapshot.Current != unlocked || rec.LastEventID != "evt-1" { + t.Fatalf("persisted record = %+v, want version 2 / unlocked / evt-1", rec) + } +} + +func TestDriveTx_BrokerAbortNaksAndDoesNotAdvanceOffset(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + seedFunded(t, m, store) + + tx := &fakeTx{} + // committed=false: the work function succeeds but End reports no commit (a + // rebalance fenced the producer). DriveTx must nak so the input is redelivered. + sub := &fakeTransactional{tx: tx, committed: false} + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + res := h(context.Background(), msg("evt-1", "c1")) + if res.Action != source.ActionNak { + t.Fatalf("action = %v, want nak on broker abort", res.Action) + } + if got, want := sub.calls, []string{"begin", "commit"}; !equalCalls(got, want) { + t.Fatalf("call order = %v, want %v", got, want) + } +} + +func TestDriveTx_EmitFailureAbortsTransaction(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + seedFunded(t, m, store) + + // The Tx fails to produce to turnstile.out, so EmitTx errors and the + // transaction aborts before the offset is committed. + tx := &fakeTx{failOn: "turnstile.out"} + sub := &fakeTransactional{tx: tx, committed: true} + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + res := h(context.Background(), msg("evt-1", "c1")) + if res.Action != source.ActionNak { + t.Fatalf("action = %v, want nak on emit failure", res.Action) + } + if got, want := sub.calls, []string{"begin", "abort"}; !equalCalls(got, want) { + t.Fatalf("call order = %v, want %v (abort on emit failure)", got, want) + } + // The transaction aborted, so nothing was committed; the seed version stands. + rec, _, _ := store.Load(context.Background(), keyOf) + if rec.Version != 1 { + t.Fatalf("version = %d, want 1 (abort persisted nothing observable; the store save was rolled back logically)", rec.Version) + } +} + +func TestDriveTx_RedeliveryIsSkip_NotTransactional(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + seedFunded(t, m, store) + + tx := &fakeTx{} + sub := &fakeTransactional{tx: tx, committed: true} + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + if first := h(context.Background(), msg("evt-1", "c1")); first.Action != source.ActionManual { + t.Fatalf("first delivery action = %v, want manual", first.Action) + } + // Redeliver the same id: deduped by version, acked as a Skip, NOT a new + // transaction. + callsBefore := len(sub.calls) + redo := h(context.Background(), msg("evt-1", "c1")) + if redo.Action != source.ActionAck || redo.Class != source.Drop { + t.Fatalf("redelivery = %v/%v, want skip (ack/drop)", redo.Action, redo.Class) + } + if len(sub.calls) != callsBefore { + t.Fatalf("redelivery opened a transaction (%v), want none", sub.calls[callsBefore:]) + } +} + +func TestDriveTx_RouteFailureIsTerm_NoTransaction(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + tx := &fakeTx{} + sub := &fakeTransactional{tx: tx, committed: true} + + boom := errors.New("cannot route") + router := func(source.Message) (turnstileState, turnstileEvent, error) { return locked, coin, boom } + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, router, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + res := h(context.Background(), msg("evt-1", "c1")) + if res.Action != source.ActionTerm { + t.Fatalf("action = %v, want term on route failure", res.Action) + } + if len(sub.calls) != 0 { + t.Fatalf("a transaction was opened on a route failure: %v", sub.calls) + } +} + +func TestDriveTx_GuardRejectionIsReject_NoTransaction(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + seed(t, m, store, &turnstile{Funded: false}) // unfunded: coin guard fails + tx := &fakeTx{} + sub := &fakeTransactional{tx: tx, committed: true} + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + + res := h(context.Background(), msg("evt-1", "c1")) + if res.Action != source.ActionTerm || res.Class != source.InvalidForState { + t.Fatalf("result = %v/%v, want reject (term/invalid_for_state)", res.Action, res.Class) + } + if len(sub.calls) != 0 { + t.Fatalf("a transaction was opened on a guard rejection: %v", sub.calls) + } +} + +func TestDriveTx_NilWiringNaks(t *testing.T) { + t.Parallel() + m := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, nil, statemachine.TxSinkFunc(emitOpenedToTopic), + ) + res := h(context.Background(), msg("evt-1", "c1")) + if !errors.Is(res.Err, statemachine.ErrNotTransactional) { + t.Fatalf("nil Transactional: err = %v, want ErrNotTransactional", res.Err) + } + + h2 := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + m, store, routeFunded, sub2(), nil, + ) + if res := h2(context.Background(), msg("evt-1", "c1")); !errors.Is(res.Err, statemachine.ErrNotTransactional) { + t.Fatalf("nil TxSink: err = %v, want ErrNotTransactional", res.Err) + } +} + +func sub2() source.Transactional { return &fakeTransactional{tx: &fakeTx{}, committed: true} } + +func equalCalls(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/source/statemachine/example_test.go b/source/statemachine/example_test.go index 04448d1..fcce9a8 100644 --- a/source/statemachine/example_test.go +++ b/source/statemachine/example_test.go @@ -54,6 +54,67 @@ func ExampleDrive() { // state: unlocked version: 2 } +// exampleTx is a tiny in-memory source.Tx for the DriveTx example: it prints the +// records produced into the transaction so the output is observable without a +// broker. +type exampleTx struct{} + +func (exampleTx) Produce(_ context.Context, records ...source.ProducedRecord) error { + for _, r := range records { + fmt.Printf("produce: %s -> %s\n", r.Topic, r.Value) + } + return nil +} + +// exampleTransactional is a tiny in-memory source.Transactional for the example: +// it runs the work function and commits, mirroring how Kafka's EOS session would +// commit the produced records and the consumed offset atomically. +type exampleTransactional struct{} + +func (exampleTransactional) Begin(ctx context.Context, _ source.Message, fn func(context.Context, source.Tx) error) error { + return fn(ctx, exampleTx{}) +} + +// ExampleDriveTx binds a turnstile statechart to a source.Handler with +// exactly-once consume-process-produce: firing the unlock transition produces the +// emitted effect AND commits the consumed offset in one transaction, so the +// handler returns Manual (the engine takes no further settle action). +func ExampleDriveTx() { + machine := buildTurnstile() + store := statemachine.NewMemStore[turnstileState, turnstileState, turnstileEvent, *turnstile]() + + seeded := machine.Cast(&turnstile{Funded: true}, state.WithInitialState[turnstileState](locked)) + _ = store.Save(context.Background(), locked, + statemachine.Record[turnstileState, turnstileEvent, *turnstile]{Snapshot: seeded.Snapshot(), Version: 1}, 0) + + router := func(source.Message) (turnstileState, turnstileEvent, error) { + return locked, coin, nil + } + // The TxSink turns each emitted effect into a record produced inside the + // transaction. + 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: []byte("opened by " + oe.By)}) + } + return nil + }) + + h := statemachine.DriveTx[turnstileState, turnstileEvent, *turnstile]( + machine, store, router, exampleTransactional{}, txSink, + ) + + res := h(context.Background(), msg("coin-1", "cursor-1")) + fmt.Println("action:", res.Action) + + rec, _, _ := store.Load(context.Background(), locked) + fmt.Println("state:", rec.Snapshot.Current, "version:", rec.Version) + + // Output: + // produce: turnstile.out -> opened by coin + // action: manual + // state: unlocked version: 2 +} + // ExampleCheckEvents validates that a consumer's accepted event union is // exhaustive against the machine's event alphabet, reporting an event the machine // can never handle (which would always be rejected as invalid-for-state).