feat: add confirmed replication barriers for hybrid write policies

.github/workflows/test.yml

@@ -98,6 +98,9 @@ jobs:
           echo "::endgroup::"
         done
 
+    - name: Verify confirmed replication survives source-process loss
+      run: go run ./scripts/replication-smoke.go
+
   coverage:
     runs-on: ubuntu-latest
     env:

CHANGELOG.md

@@ -5,10 +5,28 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [0.19.0] - 2026-09-05
+
+### Added
+
+- `NewReplicator` and generation-scoped `ReplicationPosition` values for hybrid
+  write policies. Ordinary writes replicate asynchronously; `Position` or
+  `Capture` plus `Wait` establish a destination-acknowledged prefix barrier.
+  `Confirmed` exposes progress, and restart resumes a separately scoped checkpoint.
+  Batched reads observe empty-chunk cursor advances; waiting operations wake idle
+  polling without bypassing failure backoff.
+- Strict checkpoint handling for confirmed replication: no confirmation after
+  an ambiguous append or failed checkpoint save; concurrent waits are independent,
+  cancellation does not undo writes, and generation mismatches fail explicitly.
+- Durable Streams `WithStrictDecoding()` for replication sources that must return
+  malformed-envelope errors instead of silently skipping records.
+- Confirmed Replication guide, executable API example, and a process-loss smoke
+  test that kills a SQLite source and recovers the confirmed prefix independently.
 
 ### Fixed
 
+- Make async-capacity cancellation coverage deterministic: cancel from a filter
+  after admission so both live and durable Once rollback paths are exercised.
 - Prevent `Shutdown` from closing the event store while an accepted publish,
   synchronous handler, replay, subscription setup or follower still uses it.
 - Close the event store exactly once across concurrent and repeated shutdown
@@ -902,6 +920,8 @@ Initial release of ebu (Event BUs) - a lightweight, type-safe event bus for Go.
 - `ClearAll`: Remove all handlers
 - `WaitAsync`: Wait for async handlers to complete
 
+[0.19.0]: https://github.com/jilio/ebu/compare/v0.18.1...v0.19.0
+[0.18.1]: https://github.com/jilio/ebu/compare/v0.18.0...v0.18.1
 [0.18.0]: https://github.com/jilio/ebu/compare/v0.17.0...v0.18.0
 [0.17.0]: https://github.com/jilio/ebu/compare/v0.16.0...v0.17.0
 [0.16.0]: https://github.com/jilio/ebu/compare/v0.15.0...v0.16.0

README.md

@@ -22,10 +22,14 @@ A lightweight, type-safe event bus for Go with generics support. Build decoupled
 - 💾 **Event persistence** - Built-in support for event storage and replay
 - 🔗 **Cross-process delivery** - `Follow` a shared store to span processes and machines
 - 🪞 **Log replication** - `Mirror` one store into another with envelopes preserved
+- **Confirmed replication** - `Replicator` lets important writes wait for an acknowledged prefix while ordinary writes remain asynchronous
 - 🌍 **Remote storage** - Native support for remote backends like [durable-streams](https://github.com/durable-streams/durable-streams)
 - 🔄 **Event upcasting** - Seamless event schema migration and versioning
 - ✅ **100% test coverage** - Thoroughly tested for reliability
 
+See the [Confirmed Replication Guide](docs/REPLICATION.md) for hybrid write policies,
+recovery contracts, generations and destination durability requirements.
+
 ## Installation
 
 ```bash
@@ -397,6 +401,13 @@ preserved verbatim — a local write-ahead log shipped into a shared store, a
 store migration, a rebuilt replica. See the Distributed Guide's
 "Mirroring one log into another".
 
+For hybrid durability policies, `NewReplicator` binds a source, destination and
+checkpoint store. Run it once, then use `Position` on an append result or `Capture`
+on a group of writes, followed by `Wait` for important operations. A successful
+wait inherits the destination's acknowledgement guarantees; configure an
+independent durable destination for protection against source-worker loss. See
+[Confirmed Replication](docs/REPLICATION.md).
+
 See [**Distributed Guide**](docs/DISTRIBUTED.md) for delivery modes,
 failure behavior, and current limitations.
 

async_capacity_test.go

@@ -0,0 +1,51 @@
+package eventbus
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+)
+
+// Cancellation must roll back Once when the async slot was never acquired.
+// Cancel inside the filter so it deterministically happens after the top-of-loop
+// context check and before the capacity select; scheduler timing cannot bypass
+// the branch under test.
+func TestCanceledAsyncCapacityPreservesOnceHandler(t *testing.T) {
+	for name, mode := range map[string]dispatchMode{"live": dispatchNonBlocking, "durable": dispatchWaitForAsync} {
+		t.Run(name, func(t *testing.T) {
+			bus := New(WithAsyncHandlerLimit(1))
+			bus.asyncSem <- struct{}{}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			first := true
+			calls := 0
+			err := Subscribe(bus, func(TestEvent) { calls++ }, Async(), Once(), WithFilter(func(TestEvent) bool {
+				if first {
+					first = false
+					cancel()
+				}
+				return true
+			}))
+			if err != nil {
+				t.Fatal(err)
+			}
+			e := TestEvent{}
+			err = dispatchWithMode(bus, ctx, reflect.TypeOf(e), EventType(e), e, mode)
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("canceled dispatch: %v", err)
+			}
+			<-bus.asyncSem
+			if calls != 0 || HandlerCount[TestEvent](bus) != 1 {
+				t.Fatal("unstarted Once handler was lost")
+			}
+			if err := dispatchWithMode(bus, context.Background(), reflect.TypeOf(e), EventType(e), e, mode); err != nil {
+				t.Fatal(err)
+			}
+			bus.Wait()
+			if calls != 1 {
+				t.Fatalf("retry calls = %d", calls)
+			}
+		})
+	}
+}

docs/DISTRIBUTED.md

@@ -197,6 +197,23 @@ Like durable followers, `Mirror` does not coordinate writers across
 processes: run one mirror per subscription ID, enforced by an external lease
 or a single owner.
 
+## Hybrid write policies with Replicator
+
+Use `NewReplicator` when ordinary writes may remain asynchronous but selected
+operations must wait for destination acknowledgement. It shares Mirror's copy
+loop and adds generation-scoped `ReplicationPosition` values, `Capture`, `Wait`
+and `Confirmed`. A barrier acknowledges the entire source prefix, not only the
+important event. A timeout leaves the copy running; a failed checkpoint save
+cannot advance confirmed progress.
+
+The destination's Append contract determines durability. Configure an independent
+durable destination for source-worker-loss protection. A replica acknowledgement
+is distinct from projection application or permission to promote a new writer.
+Use strict decoding on sources and preserve unreplicated history. See the
+[Confirmed Replication Guide](REPLICATION.md) for complete examples, restart and
+retention rules, and the difference between ordinary Mirror's checkpoint policy
+and confirmed replication.
+
 ## At-least-once, and what to do about it
 
 Every layer of a shared log is at-least-once: appends may be retried, chunked

docs/EXAMPLES.md

@@ -11,6 +11,7 @@ Comprehensive examples demonstrating all features of the ebu event bus library.
 - [Panic Recovery](#panic-recovery)
 - [Global Hooks](#global-hooks)
 - [Dynamic Handler Management](#dynamic-handler-management)
+- [Hybrid Replication](#hybrid-replication)
 
 ## Basic Usage
 
@@ -623,3 +624,27 @@ func (e PaymentProcessedEvent) EventTypeName() string {
 6. **Clean up handlers** - Retain subscription handles and call `sub.Unsubscribe()` when done; use `Clear` to remove all handlers of a type
 7. **Set panic handlers** - Monitor and log handler failures
 8. **Test concurrency** - The bus is thread-safe, test your handlers too
+
+## Hybrid Replication
+
+The [Confirmed Replication Guide](REPLICATION.md) includes a complete
+`appendWithReplication` helper: ordinary operations return after the local append,
+while important operations call `Replicator.Wait` for a source position. It also
+shows a barrier after `TryPublishContext` using `Capture`, and explains how to
+retry confirmation after a timeout without reissuing a write.
+
+Run the executable, in-memory API example:
+
+```sh
+go test -run '^ExampleReplicator$' .
+```
+
+For a process-loss test using SQLite and HTTP, run from the repository workspace:
+
+```sh
+go run ./scripts/replication-smoke.go
+```
+
+The smoke test kills the source process after a confirmed barrier and recovers
+its prefix from the reopened destination. Production durability still depends on
+the destination's storage configuration; MemoryStore is only a demonstration.

docs/PERSISTENCE.md

@@ -955,6 +955,26 @@ testBus.Replay(ctx, eventbus.OffsetOldest, func(event *eventbus.StoredEvent) err
 })
 ```
 
+## Replication acknowledgement and durability
+
+`Replicator` can wait until a local append's entire prefix has been acknowledged
+by another EventStore and checkpointed. This supports an asynchronous default and
+per-operation confirmed writes without changing the local write format. Its
+positions are source offsets scoped by relationship ID and log generation.
+
+A successful destination Append carries that store's guarantees: MemoryStore is
+not durable, SQLite's WAL `synchronous=NORMAL` is not a destination-power-loss
+guarantee for the latest commit, and a remote acknowledgement depends on server
+storage configuration. Use a destination outside the source worker's failure
+domain and configure its acknowledgement policy explicitly. Replication cannot
+strengthen an acknowledgement returned too early by a backend.
+
+A Durable Streams source should enable `WithStrictDecoding()` for loss-sensitive
+copying: malformed envelopes then return indexed Read/Tail errors rather than
+being skipped. Default decoding behavior is unchanged. See
+[Confirmed Replication](REPLICATION.md) for the API, checkpoint compatibility,
+timeouts, generations, retention, duplicate handling and recovery requirements.
+
 ## Shutdown and store lifetime
 
 `bus.Shutdown(ctx)` stops new bus operations, cancels active followers, and waits