Route transaction commits through configured brokers

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 |
 
 ---

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`

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() {

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 {

website/dist/docs/index.html

@@ -209,7 +209,7 @@ <h2 tabindex="-1">Features<a class="heading-anchor" href="#features" aria-label=
 <li><strong>Back-pressure policies</strong>: <code>DropOldest</code>, <code>Block</code>, or <code>ReturnError</code>.</li>
 <li><strong>Dead-letter queue</strong> with replay support and panic recovery.</li>
 <li><strong>Redis Streams and RabbitMQ providers</strong> for cross-process delivery.</li>
-<li><strong>Transactions</strong> with local buffering and synchronous commit.</li>
+<li><strong>Transactions</strong> with local buffering and transport-aware commit to local handlers or the configured broker.</li>
 <li><strong>Scheduling</strong> with absolute and relative timers.</li>
 <li><strong>Metrics</strong> for published, processed, and failed events.</li>
 <li><strong>Pluggable decision cache</strong> through the <code>DecisionCache</code> interface.</li>
@@ -671,9 +671,10 @@ <h2 tabindex="-1">Transactions<a class="heading-anchor" href="#transactions" ari
 }</code></pre></div>
 <div class="doc-flow" aria-label="Rules cache Jev decision transaction commit"><span><small>01</small>Rules / Cache / Jev</span><span><small>02</small>DecideAndPublish</span><span><small>03</small>Transaction buffer</span><span><small>04</small>Commit</span></div>
 <p><code>DecideAndPublish</code> resolves the event projection immediately and buffers only the selected event. Routing failures do not mutate the transaction, and <code>Rollback</code> discards intelligently selected events just like direct events.</p>
-<aside class="doc-note"><strong>Transactions stay local.</strong><p>Even when the parent EventStore has Redis or RabbitMQ configured, transaction commits execute buffered regular handlers synchronously. They do not publish transaction entries to the broker.</p></aside>
+<aside class="doc-note"><strong>Commit follows the EventStore transport.</strong><p>Transactions always buffer events locally until <code>Commit</code>. On a normal local store, commit executes buffered regular handlers synchronously. With <code>WithRedis(...)</code>, <code>WithRabbitMQ(...)</code>, or <code>WithProvider(...)</code>, commit publishes the buffered events to that provider instead. Local handlers run only when those broker events are consumed back through <code>store.Consume(ctx)</code>.</p></aside>
 </section>
-<p><code>Commit</code> stops at the first handler error. Transactions are not database-style atomic transactions, so handler side effects that already ran cannot be undone.</p>
+<p>Local <code>Commit</code> stops at the first handler error. Broker-backed <code>Commit</code> 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.</p>
+<p>Transactions are not database-style atomic transactions: handler side effects or broker publishes that already succeeded cannot be undone.</p>
 <p><code>Rollback</code> discards only events still buffered locally and never rewinds the shared EventStore ring.</p>
 </section>
 <section class="doc-section" id="scheduling" data-doc-section data-search-section data-title="Scheduling" data-category="Operations">
@@ -801,7 +802,7 @@ <h2 tabindex="-1">API reference<a class="heading-anchor" href="#api-snapshot" ar
 <tr><td><code>Metrics</code></td><td>Read event counters</td></tr>
 <tr><td><code>Schedule</code> / <code>ScheduleAfter</code></td><td>Schedule known events</td></tr>
 <tr><td><code>DecideAndSchedule</code> / <code>DecideAndScheduleAfter</code></td><td>Select with Rules/Cache/Jev, then schedule the resolved event</td></tr>
-<tr><td><code>BeginTransaction</code></td><td>Start a local transaction buffer; use <code>tx.DecideAndPublish</code> for Rules/Cache/Jev selection</td></tr>
+<tr><td><code>BeginTransaction</code></td><td>Buffer events until commit; local stores execute regular handlers, provider-backed stores publish to the configured broker</td></tr>
 <tr><td><code>Drain</code> / <code>Close</code></td><td>Stop accepting events and finish in-flight work</td></tr>
 </tbody></table></div>
 </section>
@@ -814,7 +815,7 @@ <h2 tabindex="-1">Examples<a class="heading-anchor" href="#examples" aria-label=
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/routing_rules"><code>routing_rules</code></a></td><td>Deterministic first-match routing with <code>EventRule</code>; no model or cache required</td></tr>
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/routing_cache"><code>routing_cache</code></a></td><td>Reusing a previous decision so the fallback selector is called only once</td></tr>
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/routing_jev"><code>routing_jev</code></a></td><td>Full <code>rules → cache → Jev → GoEventBus</code> routing with OpenRouter</td></tr>
-<tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/transaction_jev"><code>transaction_jev</code></a></td><td>Rules/Cache/Jev event selection buffered until transaction commit</td></tr>
+<tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/transaction_jev"><code>transaction_jev</code></a></td><td>Rules/Cache/Jev selection buffered until commit, then dispatched locally or published through the configured provider</td></tr>
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/schedule_jev"><code>schedule_jev</code></a></td><td>Rules/Cache/Jev event selection before delayed scheduling</td></tr>
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/hello_world"><code>hello_world</code></a></td><td>Minimal direct <code>Subscribe → Publish</code> flow</td></tr>
 <tr><td><a href="https://github.com/Protocol-Lattice/GoEventBus/tree/main/examples/middleware"><code>middleware</code></a></td><td>Handler middleware and lifecycle behavior</td></tr>