From 1f2b70a1f42ea9dd75b3c6df0c940416f3f1effd Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:16:11 +0200 Subject: [PATCH 1/6] Route transaction commits through configured providers --- transaction.go | 45 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/transaction.go b/transaction.go index 25fb446..ed6d5b3 100644 --- a/transaction.go +++ b/transaction.go @@ -15,9 +15,11 @@ import ( // implementation rewound global queue positions on Rollback and could therefore // erase events owned by concurrent producers. // -// Commit delivers buffered events synchronously and returns the first handler -// error. It does not use batch or ordered handlers, matching the historical -// transaction contract. Event selection happens before buffering; handler side +// Commit routes buffered events through the parent EventStore. When a provider +// is configured, buffered events are published to that provider in order and +// local handlers are left to the provider consumer path. Without a provider, +// regular handlers still execute synchronously, matching the historical local +// transaction contract. Event selection happens before buffering; delivery side // effects remain deferred until Commit. type Transaction struct { store *EventStore @@ -63,6 +65,11 @@ func (tx *Transaction) DecideAndPublish( // Commit delivers the events buffered when the call begins. Events appended // concurrently stay in the transaction for a later Commit. +// +// If the parent EventStore has a configured provider, Commit publishes buffered +// events to that provider instead of invoking local handlers. Confirmed broker +// publishes are removed from the transaction immediately; if a later publish +// fails, only the failed event and the remaining suffix stay buffered for retry. func (tx *Transaction) Commit(ctx context.Context) error { if ctx == nil { ctx = context.Background() @@ -79,6 +86,24 @@ func (tx *Transaction) Commit(ctx context.Context) error { events := append([]Event(nil), tx.events...) tx.mu.Unlock() + if tx.store.hasConfiguredProvider() { + provider, err := tx.store.providerForUse() + if err != nil { + return err + } + + published := 0 + for _, event := range events { + if err := provider.Publish(ctx, event); err != nil { + tx.removeCommitted(published) + return err + } + published++ + } + tx.removeCommitted(published) + return nil + } + for _, event := range events { event.Ctx = ctx atomic.AddUint64(&tx.store.publishedCount, 1) @@ -89,12 +114,20 @@ func (tx *Transaction) Commit(ctx context.Context) error { } } - tx.mu.Lock() - tx.events = tx.events[len(events):] - tx.mu.Unlock() + tx.removeCommitted(len(events)) return nil } +func (tx *Transaction) removeCommitted(count int) { + if count == 0 { + return + } + + tx.mu.Lock() + defer tx.mu.Unlock() + tx.events = tx.events[count:] +} + // Rollback discards only this transaction's local buffer. It never mutates // the shared ring, so it is safe alongside ordinary producers and publishers. func (tx *Transaction) Rollback() { From f3dd13b8063327634e85874f73bf311ab0cf7018 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:16:27 +0200 Subject: [PATCH 2/6] Test provider-backed transaction commits --- transaction_test.go | 87 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) diff --git a/transaction_test.go b/transaction_test.go index 566562b..2126680 100644 --- a/transaction_test.go +++ b/transaction_test.go @@ -28,6 +28,93 @@ func (f transactionSelectorFunc) SelectEvent( return f(ctx, state, candidates) } +type transactionRecordingProvider struct { + published []Event + failAt int +} + +func (p *transactionRecordingProvider) Publish(_ context.Context, event Event) error { + if p.failAt > 0 && len(p.published)+1 == p.failAt { + return errors.New("broker publish failed") + } + p.published = append(p.published, event) + return nil +} + +func (p *transactionRecordingProvider) Consume(context.Context, EventConsumer) error { + return nil +} + +func (p *transactionRecordingProvider) Close() error { + return nil +} + +func TestTransaction_CommitPublishesToConfiguredProvider(t *testing.T) { + var localCalls uint64 + dispatcher := Dispatcher{ + "remote": hs(makeCounterHandler(&localCalls)), + } + provider := &transactionRecordingProvider{} + store := NewEventStore(&dispatcher, 8, DropOldest, WithProvider(provider)) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + tx := store.BeginTransaction() + tx.Publish(Event{ID: "broker-1", Projection: "remote"}) + tx.Publish(Event{ID: "broker-2", Projection: "remote"}) + + if err := tx.Commit(context.Background()); err != nil { + t.Fatalf("Commit: %v", err) + } + if got := atomic.LoadUint64(&localCalls); got != 0 { + t.Fatalf("local handler calls = %d, want 0 before provider consumption", got) + } + if len(provider.published) != 2 { + t.Fatalf("provider publishes = %d, want 2", len(provider.published)) + } + if provider.published[0].ID != "broker-1" || provider.published[1].ID != "broker-2" { + t.Fatalf("provider publish order = [%q %q], want [broker-1 broker-2]", + provider.published[0].ID, provider.published[1].ID) + } + + if err := tx.Commit(context.Background()); err != nil { + t.Fatalf("second Commit: %v", err) + } + if len(provider.published) != 2 { + t.Fatalf("second Commit republished events: got %d publishes, want 2", len(provider.published)) + } +} + +func TestTransaction_CommitProviderFailureRetainsOnlyUnpublishedSuffix(t *testing.T) { + provider := &transactionRecordingProvider{failAt: 2} + store := NewEventStore(&Dispatcher{}, 8, DropOldest, WithProvider(provider)) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + tx := store.BeginTransaction() + tx.Publish(Event{ID: "broker-1", Projection: "remote"}) + tx.Publish(Event{ID: "broker-2", Projection: "remote"}) + tx.Publish(Event{ID: "broker-3", Projection: "remote"}) + + if err := tx.Commit(context.Background()); err == nil { + t.Fatal("Commit succeeded, want provider error") + } + if len(provider.published) != 1 || provider.published[0].ID != "broker-1" { + t.Fatalf("published before failure = %#v, want only broker-1", provider.published) + } + + provider.failAt = 0 + if err := tx.Commit(context.Background()); err != nil { + t.Fatalf("retry Commit: %v", err) + } + if len(provider.published) != 3 { + t.Fatalf("provider publishes after retry = %d, want 3", len(provider.published)) + } + for i, want := range []string{"broker-1", "broker-2", "broker-3"} { + if provider.published[i].ID != want { + t.Fatalf("provider publish %d = %q, want %q", i, provider.published[i].ID, want) + } + } +} + func TestTransaction_DecideAndPublishWithJev(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { From 14a15f071a3ea98270cbfa580699b28785ee3bb4 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:16:38 +0200 Subject: [PATCH 3/6] Document broker-backed transaction commits --- examples/README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/examples/README.md b/examples/README.md index 7f0cf43..68eb0f6 100644 --- a/examples/README.md +++ b/examples/README.md @@ -60,8 +60,10 @@ The example exits without making a request when `OPENROUTER_API_KEY` is not set. ### `transaction_jev` Uses the same Rules / Cache / Jev selector pipeline inside a transaction. -`DecideAndPublish` resolves the projection immediately, buffers the selected -event, and handlers run only when the transaction commits. +`DecideAndPublish` resolves the projection immediately and buffers the selected +event. On commit, local stores run regular handlers synchronously; stores created +with `WithRedis(...)`, `WithRabbitMQ(...)`, or `WithProvider(...)` publish the +buffered events to that provider instead. ```bash OPENROUTER_API_KEY=... go run ./examples/transaction_jev @@ -70,7 +72,7 @@ OPENROUTER_API_KEY=... go run ./examples/transaction_jev Flow: ```text -state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> handler +state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> local handler OR broker ``` ### `schedule_jev` From 038a77ae398c5ce6324f8e34276997a05f2543a4 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:16:41 +0200 Subject: [PATCH 4/6] Document provider-aware transaction commits --- README.MD | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/README.MD b/README.MD index 3a4262a..4cfb079 100644 --- a/README.MD +++ b/README.MD @@ -857,13 +857,17 @@ resolved event. This means: - `RuleCacheSelector`, `JevSelector`, and custom `EventSelector` implementations all use the same transaction API. -Transactions remain **local** even when the parent `EventStore` has Redis or -RabbitMQ configured. A transaction commit executes its buffered regular -handlers synchronously; it does not publish transaction entries to a broker. - -`Commit` stops at the first handler error. As before, transactions are not -database-style atomic transactions: side effects from handlers that already ran -cannot be undone. +Transaction commit follows the parent `EventStore` transport. With a normal +local store, `Commit` executes buffered regular handlers synchronously. With +`WithRedis(...)`, `WithRabbitMQ(...)`, or `WithProvider(...)`, `Commit` +publishes buffered events to the configured provider instead; local handlers run +only when those events are consumed back into an `EventStore`. + +Local `Commit` stops at the first handler error. Broker-backed `Commit` stops at +the first publish error and removes only the prefix already confirmed by the +provider, so retrying starts from the first unpublished event. Transactions are +still not database-style atomic transactions: handler side effects or broker +publishes that already succeeded cannot be undone. `Rollback` only discards events still buffered locally and never rewinds the shared EventStore ring. @@ -1055,7 +1059,7 @@ Important methods: | `Metrics` | Read event counters | | `Schedule` / `ScheduleAfter` | Schedule known events | | `DecideAndSchedule` / `DecideAndScheduleAfter` | Select with Rules/Cache/Jev, then schedule the resolved event | -| `BeginTransaction` | Start a local transaction buffer; use `tx.DecideAndPublish` for Rules/Cache/Jev selection | +| `BeginTransaction` | Buffer events for commit; local stores execute handlers, provider-backed stores publish to the broker | | `Drain` / `Close` | Stop accepting events and finish in-flight work | --- From 02fe929e7f04bef5f9ba56b809e9c26538332cc5 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:24:33 +0200 Subject: [PATCH 5/6] Update transaction docs for broker-backed commits --- website/dist/docs/index.html | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/website/dist/docs/index.html b/website/dist/docs/index.html index aaec243..8f265cb 100644 --- a/website/dist/docs/index.html +++ b/website/dist/docs/index.html @@ -209,7 +209,7 @@

FeaturesBack-pressure policies: DropOldest, Block, or ReturnError.
  • Dead-letter queue with replay support and panic recovery.
  • Redis Streams and RabbitMQ providers for cross-process delivery.
    • -
  • Transactions with local buffering and synchronous commit.
    • +
  • Transactions with local buffering and transport-aware commit to local handlers or the configured broker.
  • Scheduling with absolute and relative timers.
  • Metrics for published, processed, and failed events.
  • Pluggable decision cache through the DecisionCache interface.
    • @@ -671,9 +671,10 @@

    Transactions
    01Rules / Cache / Jev02DecideAndPublish03Transaction buffer04Commit

    DecideAndPublish resolves the event projection immediately and buffers only the selected event. Routing failures do not mutate the transaction, and Rollback discards intelligently selected events just like direct events.

    - + -

    Commit stops at the first handler error. Transactions are not database-style atomic transactions, so handler side effects that already ran cannot be undone.

    +

    Local Commit stops at the first handler error. Broker-backed Commit publishes events in transaction order, stops at the first provider error, and removes only the prefix already confirmed by the provider. Retrying the transaction therefore starts from the first unpublished event instead of publishing the confirmed prefix again.

    +

    Transactions are not database-style atomic transactions: handler side effects or broker publishes that already succeeded cannot be undone.

    Rollback discards only events still buffered locally and never rewinds the shared EventStore ring.

    @@ -801,7 +802,7 @@

    API referenceMetricsRead event counters Schedule / ScheduleAfterSchedule known events DecideAndSchedule / DecideAndScheduleAfterSelect with Rules/Cache/Jev, then schedule the resolved event -BeginTransactionStart a local transaction buffer; use tx.DecideAndPublish for Rules/Cache/Jev selection +BeginTransactionBuffer events until commit; local stores execute regular handlers, provider-backed stores publish to the configured broker Drain / CloseStop accepting events and finish in-flight work

    @@ -814,7 +815,7 @@

    Examplesrouting_rulesDeterministic first-match routing with EventRule; no model or cache required routing_cacheReusing a previous decision so the fallback selector is called only once routing_jevFull rules → cache → Jev → GoEventBus routing with OpenRouter -transaction_jevRules/Cache/Jev event selection buffered until transaction commit +transaction_jevRules/Cache/Jev selection buffered until commit, then dispatched locally or published through the configured provider schedule_jevRules/Cache/Jev event selection before delayed scheduling hello_worldMinimal direct Subscribe → Publish flow middlewareHandler middleware and lifecycle behavior From af905a2c407c1ca07c66ab076c8b0cac66b0e61a Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Sun, 4 Oct 2026 20:24:37 +0200 Subject: [PATCH 6/6] Align README transaction docs with broker-backed commits --- README.MD | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.MD b/README.MD index 4cfb079..268a987 100644 --- a/README.MD +++ b/README.MD @@ -64,7 +64,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish() - **Back-pressure policies**: `DropOldest`, `Block`, or `ReturnError`. - **Dead-letter queue** with replay support and panic recovery. - **Redis Streams and RabbitMQ EventStore options** for cross-process delivery. -- **Transactions** with local buffering, synchronous commit, and optional Rules/Cache/Jev event selection. +- **Transactions** with local buffering, transport-aware commit, and optional Rules/Cache/Jev event selection. - **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection. - **Metrics** for published, processed, and failed events. - **Pluggable decision cache** through the `DecisionCache` interface. @@ -321,7 +321,7 @@ The repository includes runnable examples for both the classic event-bus path an | [`routing_rules`](examples/routing_rules) | Deterministic first-match routing with `EventRule`; no model or cache required | | [`routing_cache`](examples/routing_cache) | Reusing a previous decision so the fallback selector is called only once | | [`routing_jev`](examples/routing_jev) | Full `rules → cache → Jev → GoEventBus` routing with OpenRouter | -| [`transaction_jev`](examples/transaction_jev) | Rules/Cache/Jev event selection buffered until transaction commit | +| [`transaction_jev`](examples/transaction_jev) | Rules/Cache/Jev selection buffered until commit, then dispatched locally or published through the configured provider | | [`schedule_jev`](examples/schedule_jev) | Rules/Cache/Jev event selection before delayed scheduling | | [`hello_world`](examples/hello_world) | Minimal direct `Subscribe → Publish` flow | | [`middleware`](examples/middleware) | Handler middleware and lifecycle behavior | @@ -776,7 +776,7 @@ Handler panics are recovered and converted into errors so a single handler canno # Transactions -Transactions are local event buffers, not database-style atomic transactions. +Transactions buffer events locally until commit, but commit delivery follows the parent `EventStore` transport. They are not database-style atomic transactions. Known events can still be buffered directly: