diff --git a/README.MD b/README.MD index 3a4262a..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: @@ -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. +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`. -`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. +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 | --- 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` 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() { 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 { 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