From 36d2885f6eea72d563033029793f19947bd2fdf5 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sat, 5 Sep 2026 19:03:49 +1000 Subject: [PATCH 1/3] feat: add confirmed replication prefix barriers --- .github/workflows/test.yml | 3 + CHANGELOG.md | 21 +- README.md | 11 + async_capacity_test.go | 51 ++ docs/DISTRIBUTED.md | 17 + docs/EXAMPLES.md | 25 + docs/PERSISTENCE.md | 20 + docs/REPLICATION.md | 212 +++++++ mirror.go | 47 +- persist.go | 4 + replication.go | 258 +++++++++ replication_example_test.go | 52 ++ replication_test.go | 640 +++++++++++++++++++++ scripts/replication-smoke.go | 213 +++++++ stores/durablestream/options.go | 10 + stores/durablestream/store.go | 9 + stores/durablestream/strict_decode_test.go | 80 +++ 17 files changed, 1665 insertions(+), 8 deletions(-) create mode 100644 async_capacity_test.go create mode 100644 docs/REPLICATION.md create mode 100644 replication.go create mode 100644 replication_example_test.go create mode 100644 replication_test.go create mode 100644 scripts/replication-smoke.go create mode 100644 stores/durablestream/strict_decode_test.go diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 47e4720..667e715 100644 --- a/.github/workflows/test.yml +++ b/.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: diff --git a/CHANGELOG.md b/CHANGELOG.md index 86a2147..81cd90b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,27 @@ 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. +- 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 +919,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 diff --git a/README.md b/README.md index c72606e..ae0a02d 100644 --- a/README.md +++ b/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. diff --git a/async_capacity_test.go b/async_capacity_test.go new file mode 100644 index 0000000..ecceb8a --- /dev/null +++ b/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) + } + }) + } +} diff --git a/docs/DISTRIBUTED.md b/docs/DISTRIBUTED.md index 2a5b53e..56716c1 100644 --- a/docs/DISTRIBUTED.md +++ b/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 diff --git a/docs/EXAMPLES.md b/docs/EXAMPLES.md index 47b5449..7f0da7a 100644 --- a/docs/EXAMPLES.md +++ b/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. diff --git a/docs/PERSISTENCE.md b/docs/PERSISTENCE.md index 7a8f21b..84b780e 100644 --- a/docs/PERSISTENCE.md +++ b/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 diff --git a/docs/REPLICATION.md b/docs/REPLICATION.md new file mode 100644 index 0000000..a8aa4f4 --- /dev/null +++ b/docs/REPLICATION.md @@ -0,0 +1,212 @@ +# Confirmed replication + +`Replicator` combines asynchronous log copying with explicit prefix barriers. +Use it when ordinary writes may return after local storage, while selected writes +must wait for an independently stored copy. It uses the same ordered copy loop as +`Mirror`; existing `Mirror` callers keep their behavior. + +## What a successful wait means + +For a position returned by `r.Position(sourceAppendOffset)` or `r.Capture(ctx)`, +`r.Wait(ctx, position)` returns nil only after the destination has acknowledged +all source events through that boundary and the source checkpoint has been saved. +An important write therefore protects its preceding ordinary writes too. Events +appended later do not move an already captured boundary. + +This is a **destination-acknowledgement guarantee**. To survive source-worker loss, +the destination must be outside that worker's failure domain. Its successful +`Append` must mean the storage durability you require: ebu cannot turn an +in-memory store, buffered write, or HTTP server that acknowledges before durable +storage into a durable replica. `MemoryStore` examples demonstrate the API only. +SQLite uses WAL with `synchronous=NORMAL`; this is not a promise that the latest +commit survives power loss on the destination. A Durable Streams acknowledgement +inherits the server/backend's configured guarantees. Do not silently downgrade +an important operation when that destination is unavailable. + +A wait does not mean a destination projection has applied the events. It also +does not acquire a lease, fence an old writer, elect a replacement worker, or +replicate snapshots. Those responsibilities belong to the application/storage +system. Source and destination offsets are different token spaces; every public +replication position contains a **source** offset. + +## Start one replication relationship + +```go +local, err := sqlite.New("documents.db") +if err != nil { return err } +remote, err := durablestream.New("https://streams.example/v1/stream", "documents-g7") +if err != nil { return err } + +r, err := eventbus.NewReplicator(eventbus.ReplicatorConfig{ + Source: local, + Destination: remote, + Checkpoints: local, + ID: "documents-to-backup", + Generation: "source-g7-destination-g3", +}, eventbus.MirrorOnError(func(err error) { log.Print(err) })) +if err != nil { return err } + +runCtx, stopReplication := context.WithCancel(context.Background()) +done := make(chan error, 1) +go func() { done <- r.Run(runCtx) }() +// Keep r, stopReplication and done in the owning service. +``` + +`NewReplicator` performs no I/O and starts no goroutine. The source must implement +`EventStoreOffsetComparer`. The bundled memory, SQLite and Durable Streams stores +do. The checkpoint store must accept the source's tokens verbatim: SQLite source +checkpoints can live in SQLite; opaque remote tokens cannot necessarily do so. +The tuple `(ID, Generation)` gets its own checkpoint namespace, separate from +ordinary `Mirror` IDs. Constructing an object does not import an older plain +Mirror checkpoint: a new relationship copies from `OffsetOldest`. + +Run exactly one owner of a relationship across processes. `Run` is single-use on +one object; concurrent or repeated calls return `ErrReplicatorStarted`. After a +process restart or a completed Run, construct another object with the same +configuration to resume. `MirrorPollInterval`, `MirrorDedupWindow`, +`MirrorOnForward` and `MirrorOnError` also configure a Replicator. Observers execute +on the copying goroutine; never wait for that replicator from an observer. + +## Mix ordinary and important writes + +Both modes use the same local append. Important operations add a wait: + +```go +func appendWithReplication( + ctx context.Context, + local eventbus.EventStore, + r *eventbus.Replicator, + event *eventbus.Event, + requireReplica bool, +) (eventbus.ReplicationPosition, error) { + offset, err := local.Append(ctx, event) + if err != nil { + return eventbus.ReplicationPosition{}, err + } + position, err := r.Position(offset) + if err != nil { + return eventbus.ReplicationPosition{}, err + } + if requireReplica { + return position, r.Wait(ctx, position) + } + return position, nil +} +``` + +The application chooses its default and can override it per operation without +restarting the copier. Assign a stable event ID before attempting the local write +and make retries of the write idempotent. If the append succeeded but the wait +timed out, keep the returned position and retry **the wait**, not the write. +An append error can also be ambiguous; that case requires the source store's +idempotency/reconciliation mechanism. + +`Position` checks the offset using the source comparer; it cannot prove that an +arbitrary token was actually issued by this log. Pass only a source Append result +whose boundary covers that append, or use `Capture`. Do not use the offset on an +arbitrary read event as evidence that the event was copied: a non-last member of +a chunk can carry the chunk's start token. The copier confirms the end only after +the complete chunk is acknowledged. + +To confirm a group of successful writes, including publishes through EventBus: + +```go +// Check every local write's result before creating the barrier. +if err := eventbus.TryPublishContext(bus, ctx, DocumentSaved{ID: "doc-1"}); err != nil { + return err +} +position, err := r.Capture(ctx) +if err != nil { return err } +waitCtx, cancel := context.WithTimeout(ctx, 5*time.Second) +defer cancel() +if err := r.Wait(waitCtx, position); err != nil { return err } +``` + +The bus's store must be this replicator's source. `Capture` resolves the current +tail once and performs no waiting. Concurrent writes can be included up to that +resolution, but writes after it cannot extend the barrier. To switch an +application default from asynchronous to confirmed, establish the required +boundary, wait for it, and use the confirmed write path for subsequent requests. +Coordinate that policy transition with producers in the application. Switching +cannot recover already lost data. + +## Timeouts, failures and progress + +| Situation | Behavior | +|---|---| +| Wait canceled or timed out | That waiter returns its context error. Copying and other waiters continue; writes are not rolled back. | +| Destination append fails, including a lost response after commit | Retry that event in place. Do not confirm it on the failed attempt. Duplicate IDs can reach the destination. | +| First checkpoint save fails | Run fails and wakes unfulfilled waiters; check token compatibility/configuration. | +| Later checkpoint save fails | Retry that save in place. Do not copy the next event or advance confirmed progress. | +| Source read/Tail fails | Report and retry without acknowledging a gap. Permanent errors can require operator recovery. | +| Run stops before the requested boundary | Wait returns an error matching `ErrReplicationStopped` and the Run error via `errors.Is`. | +| Requested boundary was already confirmed | Wait can still succeed after Run stops, subject to the same storage/retention contract. | +| Position belongs to another ID or generation | Fail immediately with `ErrReplicationPosition`. | + +`r.Confirmed()` returns the latest confirmed source position and a boolean. +The boolean is false until startup has established progress. The empty prefix is +valid. At startup a checkpoint from the same relationship/generation is trusted: +it certifies earlier destination acknowledgements. Later progress is announced +only after another successful checkpoint save. Reusing a checkpoint with a +replacement or truncated destination violates that assumption. + +Positions are JSON-serializable (`id`, `generation`, `offset`). Many callers may +wait on different positions concurrently, including before Run starts. A wait +never starts Run for you. Bound waits with a context and observe the Run result. + +## Generations, retention and recovery + +An ID identifies one source/destination relationship. A generation identifies its +immutable log incarnations. Change the generation after replacing, rebuilding or +restoring **either** log, even when its offsets happen to look the same. Old +positions are rejected and the new checkpoint namespace starts from the beginning. +Do not hot-swap either store behind a running object. Generation values are +application-owned; a new random value on every ordinary restart would needlessly +recopy the entire history. + +A saved checkpoint ahead of the source tail fails startup with +`ErrReplicationHistory`. `MirrorResetOnSourceRewind` is rejected for Replicator: +automatically resetting an acknowledged generation would make old promises +misleading. Offsets cannot detect a replaced log that has already refilled past +its old tail; assigning a new generation is mandatory in that case too. + +Keep every unreplicated source event. A fresh relationship requires the entire +source history. Do not compact/truncate it underneath copying. Keep acknowledged +destination history, or an application-managed snapshot that preserves the same +recovery guarantee. Replicator does not establish or validate snapshot coverage. +Sources must return errors for missing required history and must not filter or +silently skip records. For a Durable Streams **source**, use +`durablestream.WithStrictDecoding()` so malformed envelopes stop Read/Tail rather +than disappear from the replicated prefix. That option is unnecessary merely +because Durable Streams is the destination. + +Delivery is at-least-once. A crash after destination append but before checkpoint +save causes redelivery, possibly well outside a bounded in-memory dedup window. +Recover with durable event-ID deduplication or version-aware reducers that reject +stale replay. Reapplying an old assignment after a newer one can regress state; +a plain overwrite is not enough to handle every duplicate ordering. Only expose +a recovered projection when its required history has been applied. + +For planned handoff: stop/join producers, capture and confirm the final boundary, +cancel and join Run, then close its stores (or shut down their owning bus). +For unexpected worker loss: recover from the independent destination, accepting +that the asynchronous suffix may be absent. Granting the new worker exclusive +write authority and preventing the old worker from resuming are application +responsibilities. This API does not promise zero-downtime failover. + +## Verification + +The core tests inject blocked and ambiguous appends, failed checkpoint saves, +missing-history errors, cancellation, chunk boundaries, restart and generation +changes. The executable Go example is in `replication_example_test.go`. + +From this repository's Go workspace, run the process-loss smoke test: + +```sh +go run ./scripts/replication-smoke.go +``` + +It mirrors a child process's SQLite log over HTTP into a separate SQLite store, +waits for 201 revisions, kills the source process, removes its files, and checks +the full confirmed prefix using only the reopened destination. This exercises +source-process loss, not simultaneous destination power loss or worker fencing. diff --git a/mirror.go b/mirror.go index 2c48bd7..7a6183a 100644 --- a/mirror.go +++ b/mirror.go @@ -15,6 +15,7 @@ type mirrorConfig struct { resetOnRewind bool onForward func(*StoredEvent) onError func(error) + progress *Replicator } // MirrorPollInterval sets how long the mirror sleeps between Read calls when @@ -202,6 +203,10 @@ func Mirror(ctx context.Context, src, dst EventStore, subscriptionID string, off } } + return mirrorRun(ctx, src, dst, subscriptionID, offsets, cfg) +} + +func mirrorRun(ctx context.Context, src, dst EventStore, subscriptionID string, offsets SubscriptionStore, cfg *mirrorConfig) error { // The comparer is used opportunistically whenever the source can order // offsets (the checkpoint monotonicity guard in advance); the rewind // option additionally requires it. @@ -252,6 +257,12 @@ func Mirror(ctx context.Context, src, dst EventStore, subscriptionID string, off saveProven: found, } + if cfg.progress != nil { + if err := cfg.progress.initialize(ctx, from); err != nil { + return err + } + } + // A rebuilt source is detectable before the first read. Like every later // reconcile, the check is best-effort: a transient source error here is // reported, not fatal — the same error one loop iteration later would be @@ -469,26 +480,48 @@ func (m *mirror) forward(ctx context.Context, stored *StoredEvent) error { // with a lower chunk-start token — persisting the lower one would re-append // an already checkpointed range after a crash. // -// SaveOffset failures after the first proven save are reported and do not -// stop the mirror (the position is redundant with the events themselves). +// For plain Mirror, SaveOffset failures after the first proven save are reported +// and do not stop copying. Replicator instead retries the save in place before +// copying any further event or announcing confirmed progress. // A failure of the FIRST save is fatal: it is the signature of a checkpoint // store that cannot store the source's offset tokens at all, and continuing // would look healthy while every restart re-copies the whole source. func (m *mirror) advance(ctx context.Context, offset Offset) error { - if m.comparer != nil && m.offset != OffsetOldest { - if cmp, err := m.comparer.CompareOffsets(offset, m.offset); err == nil && cmp <= 0 { + if m.comparer != nil && (m.offset != OffsetOldest || m.cfg.progress != nil) { + cmp, err := m.comparer.CompareOffsets(offset, m.offset) + if err != nil && m.cfg.progress != nil { + return fmt.Errorf("replication: compare progress: %w", err) + } + if err == nil && cmp <= 0 { return nil } } - m.offset = offset - if err := m.offsets.SaveOffset(ctx, m.subscription, offset); err != nil { + // Plain Mirror retains its best-effort checkpoint policy. Confirmed + // replication cannot advertise progress until the checkpoint save succeeds. + if m.cfg.progress == nil { + m.offset = offset + } + for { + err := m.offsets.SaveOffset(ctx, m.subscription, offset) + if err == nil { + break + } if !m.saveProven { return fmt.Errorf("eventbus: mirror %q: initial checkpoint save failed (does the offsets store accept the source's offset tokens?): %w", m.subscription, err) } m.report(fmt.Errorf("mirror: save offset for %q: %w", m.subscription, err)) - return nil + if m.cfg.progress == nil { + return nil + } + if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil { + return err + } } + m.offset = offset m.saveProven = true + if m.cfg.progress != nil { + m.cfg.progress.confirm(offset) + } return nil } diff --git a/persist.go b/persist.go index f2928af..e8f261b 100644 --- a/persist.go +++ b/persist.go @@ -53,6 +53,10 @@ type EventStore interface { // error does not prove the event is absent: a remote commit may succeed // before its acknowledgement is lost. Implementations should use Event.ID // to make internal retries idempotent. + // A replication barrier made from this offset assumes the returned boundary + // covers the appended event; stores must not return an earlier chunk-start + // boundary for Append. Durability of a successful Append is store-specific: + // replication cannot strengthen an in-memory or buffered acknowledgement. Append(ctx context.Context, event *Event) (Offset, error) // Read returns events starting after the given offset. diff --git a/replication.go b/replication.go new file mode 100644 index 0000000..51e91ba --- /dev/null +++ b/replication.go @@ -0,0 +1,258 @@ +package eventbus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" +) + +var ( + // ErrReplicationPosition identifies an invalid position or a position from a + // different replication identity/generation. + ErrReplicationPosition = errors.New("eventbus: invalid replication position") + // ErrReplicationHistory means a saved checkpoint is ahead of the source tail. + // Restore the matching source or start a new generation; do not reuse offsets. + ErrReplicationHistory = errors.New("eventbus: replication history changed") + // ErrReplicatorStarted means Run was already called on this object. + ErrReplicatorStarted = errors.New("eventbus: replicator already started") + // ErrReplicationStopped means Run ended before the requested point was confirmed. + ErrReplicationStopped = errors.New("eventbus: replication stopped") +) + +// ReplicationPosition identifies a SOURCE boundary, scoped to one immutable +// source/destination pair and generation. It can be serialized across restarts. +// Use Replicator.Position for an Append result or Capture for a group of writes. +// Never use a non-last chunk member's StoredEvent.Offset as proof of that +// member's replication: such offsets may refer to the start of the chunk. +type ReplicationPosition struct { + ID string `json:"id"` + Generation string `json:"generation"` + Offset Offset `json:"offset"` +} + +// ReplicatorConfig binds one replication relationship. ID must identify the +// source/destination pair. Generation is an application-owned immutable epoch: +// change it after replacing, rebuilding or restoring EITHER log. Epoch changes +// require a new Replicator and checkpoint namespace; they cannot be inferred +// from opaque offsets, especially after a rebuilt log has caught up again. +// +// Source must implement EventStoreOffsetComparer, expose every event without +// filtering/skipping decode failures, and retain all history not yet replicated. A fresh generation starts at OffsetOldest and requires the +// full source history. Source and destination must be different logs with no +// cyclic mirror topology. Destination must preserve all acknowledged records (or +// a recovery-equivalent snapshot). There must be one owner of the checkpoint +// namespace across processes; Replicator does not acquire distributed leases. +// +// Wait inherits Destination.Append's acknowledgement guarantees. For survival +// of source-worker loss, Destination must be outside that worker's failure +// domain and acknowledge only after meeting your storage durability policy. +// An in-memory or buffered destination does not become durable through this API. +// Checkpoints must durably store SOURCE offsets verbatim; losing a checkpoint +// causes redelivery. Both replicas must tolerate duplicate event IDs on restart. +type ReplicatorConfig struct { + Source EventStore + Destination EventStore + Checkpoints SubscriptionStore + ID string + Generation string +} + +// Replicator runs Mirror's ordered copy loop and exposes acknowledged prefix +// barriers. Construct with NewReplicator. Run is single-use; after it ends, +// construct another instance with the same config to resume the checkpoint. +// It owns no store lifetimes and is independent of EventBus.Shutdown. +type Replicator struct { + config ReplicatorConfig + comparer EventStoreOffsetComparer + mirrorConfig *mirrorConfig + checkpointID string + mu sync.Mutex + changed chan struct{} + started bool + stopped bool + runErr error + confirmed Offset + ready bool +} + +// NewReplicator prepares a replicator without I/O or background goroutines. +// Mirror options configure retry timing, deduplication and observers. Automatic +// rewind reset is rejected: reusing an acknowledged generation after history +// replacement would invalidate its barriers. Ordinary Mirror is unchanged. +func NewReplicator(config ReplicatorConfig, opts ...MirrorOption) (*Replicator, error) { + if config.Source == nil || config.Destination == nil || config.Checkpoints == nil { + return nil, fmt.Errorf("eventbus: replication requires source, destination and checkpoint stores") + } + if config.ID == "" || config.Generation == "" { + return nil, fmt.Errorf("eventbus: replication ID and generation are required") + } + comparer, ok := config.Source.(EventStoreOffsetComparer) + if !ok { + return nil, fmt.Errorf("eventbus: replication source must implement EventStoreOffsetComparer") + } + cfg := &mirrorConfig{pollInterval: 200 * time.Millisecond, dedupWindow: 1024} + for _, opt := range opts { + if opt == nil { + return nil, fmt.Errorf("eventbus: mirror option cannot be nil") + } + if err := opt(cfg); err != nil { + return nil, err + } + } + if cfg.resetOnRewind { + return nil, fmt.Errorf("eventbus: confirmed replication cannot reset history; use a new generation") + } + // A structured tuple prevents IDs/generations containing separators from + // aliasing another relationship's checkpoints. Plain Mirror IDs stay separate. + key, _ := json.Marshal([2]string{config.ID, config.Generation}) + r := &Replicator{config: config, comparer: comparer, mirrorConfig: cfg, + checkpointID: "ebu:replication:" + string(key), changed: make(chan struct{})} + cfg.progress = r + return r, nil +} + +// Position scopes an offset returned by Source.Append to this relationship. +// It delegates token comparisons to the store; this is not an existence or +// provenance check. Pass only offsets actually issued by this source generation. OffsetNewest is rejected; use Capture to resolve it. +func (r *Replicator) Position(offset Offset) (ReplicationPosition, error) { + if offset == OffsetNewest { + return ReplicationPosition{}, fmt.Errorf("%w: newest is symbolic", ErrReplicationPosition) + } + cmp, err := r.comparer.CompareOffsets(offset, OffsetOldest) + if err != nil { + return ReplicationPosition{}, fmt.Errorf("%w: %w", ErrReplicationPosition, err) + } + if cmp < 0 { + return ReplicationPosition{}, fmt.Errorf("%w: offset precedes the start", ErrReplicationPosition) + } + return ReplicationPosition{ID: r.config.ID, Generation: r.config.Generation, Offset: offset}, nil +} + +// Capture resolves the current source tail once. Call it after successful local +// writes, then Wait for the returned position to establish a prefix barrier. +// Concurrent later appends do not extend that barrier. This method does not +// itself wait for replication or start Run. +func (r *Replicator) Capture(ctx context.Context) (ReplicationPosition, error) { + if err := ctx.Err(); err != nil { + return ReplicationPosition{}, err + } + events, offset, err := r.config.Source.Read(ctx, OffsetNewest, 0) + if err != nil { + return ReplicationPosition{}, fmt.Errorf("replication: capture source tail: %w", err) + } + if err := ctx.Err(); err != nil { + return ReplicationPosition{}, err + } + if len(events) != 0 { + return ReplicationPosition{}, fmt.Errorf("replication: tail lookup returned historical events") + } + return r.Position(offset) +} + +// Run copies until canceled or a fatal startup/checkpoint error. Transient +// mirror errors are retried in place. Concurrent/repeated Run calls fail with +// ErrReplicatorStarted; use a new instance to resume after this call returns. +// Observers run on this goroutine: never call Wait on this replicator from one. +func (r *Replicator) Run(ctx context.Context) (err error) { + r.mu.Lock() + if r.started { + r.mu.Unlock() + return ErrReplicatorStarted + } + r.started = true + r.mu.Unlock() + defer func() { + r.mu.Lock() + r.stopped = true + r.runErr = errors.Join(ErrReplicationStopped, err) + close(r.changed) + r.mu.Unlock() + }() + return mirrorRun(ctx, r.config.Source, r.config.Destination, r.checkpointID, r.config.Checkpoints, r.mirrorConfig) +} + +// Confirmed returns the latest confirmed SOURCE prefix and whether startup has +// established one. The empty prefix is valid. A previously confirmed prefix +// remains confirmed after Run stops, subject to the configured storage contract. +func (r *Replicator) Confirmed() (ReplicationPosition, bool) { + r.mu.Lock() + defer r.mu.Unlock() + return ReplicationPosition{ID: r.config.ID, Generation: r.config.Generation, Offset: r.confirmed}, r.ready +} + +// Wait waits until every source event through target has been acknowledged by +// Destination.Append and its SOURCE checkpoint has been saved. It never treats +// the local append, a read attempt, or an observer callback as acknowledgement. +// This is a replication barrier, not evidence that a destination projection has +// applied the events. Configure destination durability separately (see config). +// +// Wait may start before Run, and multiple waiters are independent. Canceling a +// wait neither cancels replication nor rolls back any writes: a timeout has an +// ambiguous outcome, so retry the same position instead of reissuing the write. +// Positions from another ID/generation fail immediately. If Run ends before +// target is confirmed, the error matches ErrReplicationStopped and Run's error. +func (r *Replicator) Wait(ctx context.Context, target ReplicationPosition) error { + if target.ID != r.config.ID || target.Generation != r.config.Generation { + return fmt.Errorf("%w: ID or generation mismatch", ErrReplicationPosition) + } + if _, err := r.Position(target.Offset); err != nil { + return err + } + for { + if err := ctx.Err(); err != nil { + return err + } + r.mu.Lock() + ready, confirmed, stopped, runErr, changed := r.ready, r.confirmed, r.stopped, r.runErr, r.changed + r.mu.Unlock() + if ready { + cmp, err := r.comparer.CompareOffsets(confirmed, target.Offset) + if err != nil { + return fmt.Errorf("replication: compare wait target: %w", err) + } + if cmp >= 0 { + return nil + } + } + if stopped { + return runErr + } + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func (r *Replicator) initialize(ctx context.Context, from Offset) error { + point, err := r.Position(from) + if err != nil { + return fmt.Errorf("replication: invalid saved checkpoint: %w", err) + } + tail, err := r.Capture(ctx) + if err != nil { + return err + } + cmp, err := r.comparer.CompareOffsets(point.Offset, tail.Offset) + if err != nil { + return fmt.Errorf("replication: compare saved checkpoint: %w", err) + } + if cmp > 0 { + return fmt.Errorf("%w: checkpoint %q is ahead of tail %q", ErrReplicationHistory, from, tail.Offset) + } + r.confirm(from) + return nil +} + +func (r *Replicator) confirm(offset Offset) { + r.mu.Lock() + r.confirmed = offset + r.ready = true + close(r.changed) + r.changed = make(chan struct{}) + r.mu.Unlock() +} diff --git a/replication_example_test.go b/replication_example_test.go new file mode 100644 index 0000000..33f752f --- /dev/null +++ b/replication_example_test.go @@ -0,0 +1,52 @@ +package eventbus_test + +import ( + "context" + "encoding/json" + "fmt" + "time" + + eventbus "github.com/jilio/ebu" +) + +func ExampleReplicator() { + // MemoryStore demonstrates the API only. Use an independent durable + // destination to protect against loss of the source worker. + local, backup := eventbus.NewMemoryStore(), eventbus.NewMemoryStore() + r, err := eventbus.NewReplicator(eventbus.ReplicatorConfig{ + Source: local, Destination: backup, Checkpoints: local, + ID: "documents-to-backup", Generation: "documents-v1-backup-v1", + }, eventbus.MirrorPollInterval(time.Millisecond)) + if err != nil { + panic(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- r.Run(ctx) }() + defer func() { cancel(); <-done }() + + // Ordinary writes return after the local append. + if _, err := local.Append(ctx, &eventbus.Event{ID: "draft-1", Type: "document", Data: json.RawMessage(`{"text":"draft"}`)}); err != nil { + panic(err) + } + // A critical write waits for the whole prefix, including that draft. + offset, err := local.Append(ctx, &eventbus.Event{ID: "saved-2", Type: "document", Data: json.RawMessage(`{"text":"saved"}`)}) + if err != nil { + panic(err) + } + point, err := r.Position(offset) + if err != nil { + panic(err) + } + waitCtx, stopWait := context.WithTimeout(ctx, time.Second) + defer stopWait() + if err := r.Wait(waitCtx, point); err != nil { + panic(err) + } + events, _, err := backup.Read(ctx, eventbus.OffsetOldest, 0) + if err != nil { + panic(err) + } + fmt.Println("confirmed events:", len(events)) + // Output: confirmed events: 2 +} diff --git a/replication_test.go b/replication_test.go new file mode 100644 index 0000000..cd9993e --- /dev/null +++ b/replication_test.go @@ -0,0 +1,640 @@ +package eventbus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "iter" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func replicationConfig(src, dst EventStore) ReplicatorConfig { + return ReplicatorConfig{Source: src, Destination: dst, Checkpoints: NewMemoryStore(), ID: "app-backup", Generation: "epoch-1"} +} +func newTestReplicator(t *testing.T, cfg ReplicatorConfig, opts ...MirrorOption) *Replicator { + t.Helper() + r, err := NewReplicator(cfg, append([]MirrorOption{MirrorPollInterval(time.Millisecond)}, opts...)...) + if err != nil { + t.Fatal(err) + } + return r +} +func startReplicator(t *testing.T, r *Replicator) (context.CancelFunc, <-chan error) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- r.Run(ctx) }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("replicator did not stop") + } + }) + return cancel, done +} +func appendReplicationEvent(t *testing.T, s EventStore, id string) Offset { + t.Helper() + o, err := s.Append(context.Background(), &Event{ID: id, Type: "replicated", Data: json.RawMessage(`{"value":1}`)}) + if err != nil { + t.Fatal(err) + } + return o +} +func replicationPoint(t *testing.T, r *Replicator, o Offset) ReplicationPosition { + t.Helper() + p, err := r.Position(o) + if err != nil { + t.Fatal(err) + } + return p +} +func TestReplicatorValidation(t *testing.T) { + src, dst := NewMemoryStore(), NewMemoryStore() + for _, name := range []string{"source", "destination", "checkpoints", "id", "generation", "comparer", "nil option", "bad option", "rewind"} { + t.Run(name, func(t *testing.T) { + c := replicationConfig(src, dst) + var opts []MirrorOption + switch name { + case "source": + c.Source = nil + case "destination": + c.Destination = nil + case "checkpoints": + c.Checkpoints = nil + case "id": + c.ID = "" + case "generation": + c.Generation = "" + case "comparer": + c.Source = &mirrorScriptStore{} + case "nil option": + opts = []MirrorOption{nil} + case "bad option": + opts = []MirrorOption{MirrorPollInterval(0)} + case "rewind": + opts = []MirrorOption{MirrorResetOnSourceRewind()} + } + if _, err := NewReplicator(c, opts...); err == nil { + t.Fatal("accepted invalid config") + } + }) + } + r := newTestReplicator(t, replicationConfig(src, dst)) + if _, ready := r.Confirmed(); ready { + t.Fatal("confirmed before startup") + } + for _, o := range []Offset{OffsetNewest} { + if _, err := r.Position(o); !errors.Is(err, ErrReplicationPosition) { + t.Fatalf("Position(%q): %v", o, err) + } + } + p := replicationPoint(t, r, OffsetOldest) + for _, bad := range []ReplicationPosition{{ID: "other", Generation: p.Generation}, {ID: p.ID, Generation: "other"}, {ID: p.ID, Generation: p.Generation, Offset: OffsetNewest}} { + if err := r.Wait(context.Background(), bad); !errors.Is(err, ErrReplicationPosition) { + t.Fatalf("foreign position: %v", err) + } + } + // Structurally distinct identities must not share checkpoint keys. + c1, c2 := replicationConfig(src, dst), replicationConfig(src, dst) + c1.ID = "a:b" + c1.Generation = "c" + c2.ID = "a" + c2.Generation = "b:c" + if newTestReplicator(t, c1).checkpointID == newTestReplicator(t, c2).checkpointID { + t.Fatal("checkpoint identity collision") + } +} + +func TestReplicationBarrierWaitsForWholePrefix(t *testing.T) { + src, replica := NewMemoryStore(), NewMemoryStore() + entered, release := make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + dst := &mirrorScriptStore{appendFn: func(ctx context.Context, e *Event) (Offset, error) { + if e.ID == "ordinary" { + close(entered) + select { + case <-release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + return replica.Append(ctx, e) + }} + cfg := replicationConfig(src, dst) + r := newTestReplicator(t, cfg) + appendReplicationEvent(t, src, "ordinary") + important := appendReplicationEvent(t, src, "important") + target := replicationPoint(t, r, important) + waiting := make(chan error, 1) + waitCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { waiting <- r.Wait(waitCtx, target) }() // Waiting before Run is supported. + startReplicator(t, r) + lifecycleReceive(t, entered) + if p, ok := r.Confirmed(); !ok || p.Offset != OffsetOldest { + t.Fatalf("premature confirmed prefix: %+v, %v", p, ok) + } + cancel() + if err := lifecycleReceive(t, waiting); !errors.Is(err, context.Canceled) { + t.Fatalf("wait cancellation: %v", err) + } + // Canceling one waiter does not cancel copying or unrelated local writes. + later := appendReplicationEvent(t, src, "later") + unblock() + if err := r.Wait(lifecycleContext(t), target); err != nil { + t.Fatal(err) + } + if err := r.Wait(lifecycleContext(t), replicationPoint(t, r, later)); err != nil { + t.Fatal(err) + } + got, _, err := replica.Read(context.Background(), OffsetOldest, 0) + if err != nil { + t.Fatal(err) + } + for i, id := range []string{"ordinary", "important", "later"} { + if got[i].ID != id { + t.Fatalf("prefix order: %+v", got) + } + } + // Confirmation survives Run stopping and serialization of the position. + raw, _ := json.Marshal(target) + var restored ReplicationPosition + if err := json.Unmarshal(raw, &restored); err != nil { + t.Fatal(err) + } + if err := r.Wait(lifecycleContext(t), restored); err != nil { + t.Fatal(err) + } +} + +func TestReplicationCaptureIsFixedAndWaitersIndependent(t *testing.T) { + src, replica := NewMemoryStore(), NewMemoryStore() + dst := &mirrorScriptStore{appendFn: func(ctx context.Context, e *Event) (Offset, error) { + if e.ID == "later" { + <-ctx.Done() + return "", ctx.Err() + } + return replica.Append(ctx, e) + }} + r := newTestReplicator(t, replicationConfig(src, dst)) + appendReplicationEvent(t, src, "included") + target, err := r.Capture(context.Background()) + if err != nil { + t.Fatal(err) + } + appendReplicationEvent(t, src, "later") + startReplicator(t, r) + var wg sync.WaitGroup + for range 16 { + wg.Add(1) + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := r.Wait(ctx, target); err != nil { + t.Error(err) + } + }() + } + wg.Wait() + if n := mirrorCount(t, replica); n != 1 { + t.Fatalf("replicated %d events", n) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := r.Wait(ctx, target); !errors.Is(err, context.Canceled) { + t.Fatalf("already canceled wait: %v", err) + } +} + +func TestReplicationAmbiguousAppendRetriesWithoutAcknowledging(t *testing.T) { + src, replica := NewMemoryStore(), NewMemoryStore() + var attempts atomic.Int32 + entered, release := make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + dst := &mirrorScriptStore{appendFn: func(ctx context.Context, e *Event) (Offset, error) { + n := attempts.Add(1) + if n == 2 { + close(entered) + select { + case <-release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + o, err := replica.Append(ctx, e) + if err != nil { + return "", err + } + if n == 1 { + return "", errors.New("acknowledgement lost after commit") + } + return o, nil + }} + r := newTestReplicator(t, replicationConfig(src, dst)) + target := replicationPoint(t, r, appendReplicationEvent(t, src, "stable-id")) + startReplicator(t, r) + lifecycleReceive(t, entered) + if p, _ := r.Confirmed(); p.Offset != OffsetOldest { + t.Fatal("ambiguous append was confirmed") + } + unblock() + if err := r.Wait(lifecycleContext(t), target); err != nil { + t.Fatal(err) + } + got, _, _ := replica.Read(context.Background(), OffsetOldest, 0) + if len(got) != 2 || got[0].ID != got[1].ID { + t.Fatalf("expected at-least-once copies with stable IDs: %+v", got) + } +} + +func TestReplicationCheckpointFailureCannotConfirmOrSkip(t *testing.T) { + src, dst := NewMemoryStore(), NewMemoryStore() + offsets := newMirrorLegacyOffsets() + cfg := replicationConfig(src, dst) + cfg.Checkpoints = offsets + failed := make(chan struct{}, 1) + // Seed a proven checkpoint so a later transient save is retried in place. + r := newTestReplicator(t, cfg, MirrorOnError(func(error) { + select { + case failed <- struct{}{}: + default: + } + })) + first := appendReplicationEvent(t, src, "first") + offsets.offsets[r.checkpointID] = first + appendReplicationEvent(t, dst, "first") + offsets.savesBeforeErr = 0 + offsets.saveErr = errors.New("checkpoint unavailable") + second := appendReplicationEvent(t, src, "second") + third := appendReplicationEvent(t, src, "third") + startReplicator(t, r) + lifecycleReceive(t, failed) + if p, _ := r.Confirmed(); p.Offset != first { + t.Fatalf("uncheckpointed progress: %+v", p) + } + if mirrorCount(t, dst) != 2 { + t.Fatal("advanced past the uncheckpointed event") + } + offsets.mu.Lock() + offsets.saveErr = nil + offsets.mu.Unlock() + if err := r.Wait(lifecycleContext(t), replicationPoint(t, r, third)); err != nil { + t.Fatal(err) + } + if n := mirrorCount(t, dst); n != 3 { + t.Fatalf("retry re-appended an already acknowledged event: %d", n) + } + if err := r.Wait(lifecycleContext(t), replicationPoint(t, r, second)); err != nil { + t.Fatal(err) + } +} + +func TestReplicationRestartAndGenerationIsolation(t *testing.T) { + src, dst := NewMemoryStore(), NewMemoryStore() + cfg := replicationConfig(src, dst) + r := newTestReplicator(t, cfg) + target := replicationPoint(t, r, appendReplicationEvent(t, src, "first")) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- r.Run(ctx) }() + if err := r.Wait(lifecycleContext(t), target); err != nil { + t.Fatal(err) + } + cancel() + if err := lifecycleReceive(t, done); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if err := r.Run(context.Background()); !errors.Is(err, ErrReplicatorStarted) { + t.Fatal(err) + } + if err := r.Wait(context.Background(), target); err != nil { + t.Fatalf("confirmed prefix invalidated by stop: %v", err) + } + future := replicationPoint(t, r, appendReplicationEvent(t, src, "second")) + if err := r.Wait(context.Background(), future); !errors.Is(err, ErrReplicationStopped) || !errors.Is(err, context.Canceled) { + t.Fatalf("stopped wait: %v", err) + } + resumed := newTestReplicator(t, cfg) + startReplicator(t, resumed) + if err := resumed.Wait(lifecycleContext(t), future); err != nil { + t.Fatal(err) + } + if mirrorCount(t, dst) != 2 { + t.Fatal("persisted prefix was recopied") + } + // A replacement destination gets a fresh epoch/checkpoint and full replay. + replacement := NewMemoryStore() + cfg.Destination = replacement + cfg.Generation = "epoch-2" + next := newTestReplicator(t, cfg) + if err := next.Wait(context.Background(), future); !errors.Is(err, ErrReplicationPosition) { + t.Fatal(err) + } + startReplicator(t, next) + if err := next.Wait(lifecycleContext(t), replicationPoint(t, next, future.Offset)); err != nil { + t.Fatal(err) + } + if mirrorCount(t, replacement) != 2 { + t.Fatal("new epoch skipped old checkpoint prefix") + } +} + +func TestReplicationStartupFailuresWakeWaiters(t *testing.T) { + for _, kind := range []string{"load", "newest", "invalid", "rewind", "first save", "canceled"} { + t.Run(kind, func(t *testing.T) { + src, dst := NewMemoryStore(), NewMemoryStore() + cfg := replicationConfig(src, dst) + offsets := newMirrorLegacyOffsets() + cfg.Checkpoints = offsets + realOffset := appendReplicationEvent(t, src, "one") + r := newTestReplicator(t, cfg) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + switch kind { + case "load": + offsets.loadErr = errors.New("load failed") + case "newest": + offsets.offsets[r.checkpointID] = OffsetNewest + case "invalid": + offsets.offsets[r.checkpointID] = "bogus" + case "rewind": + offsets.offsets[r.checkpointID] = "00000000000000000009" + case "first save": + offsets.saveErr = errors.New("cannot save offset") + case "canceled": + cancel() + } + target := replicationPoint(t, r, realOffset) + wait := make(chan error, 1) + go func() { wait <- r.Wait(context.Background(), target) }() + err := r.Run(ctx) + if err == nil { + t.Fatal("expected Run failure") + } + if kind == "rewind" && !errors.Is(err, ErrReplicationHistory) { + t.Fatalf("rewind: %v", err) + } + if err := lifecycleReceive(t, wait); !errors.Is(err, ErrReplicationStopped) { + t.Fatalf("wait: %v", err) + } + }) + } +} + +func TestReplicationChunkBoundaryNeverConfirmsUnfinishedSuffix(t *testing.T) { + for _, tailing := range []bool{false, true} { + t.Run(fmt.Sprint(tailing), func(t *testing.T) { + // The concrete target "02" falls inside the indivisible 00..03 chunk. + batch := []*StoredEvent{{ID: "a", Offset: OffsetOldest}, {ID: "b", Offset: OffsetOldest}, {ID: "c", Offset: "03"}} + read := func(_ context.Context, from Offset, _ int) ([]*StoredEvent, Offset, error) { + if from == OffsetNewest { + return nil, "03", nil + } + if from == OffsetOldest { + return batch, "03", nil + } + return nil, from, nil + } + compare := func(l, r Offset) (int, error) { return strings.Compare(string(l), string(r)), nil } + var src EventStore = &mirrorComparerStore{mirrorScriptStore: mirrorScriptStore{readFn: read}, compareFn: compare} + if tailing { + src = &mirrorTailerComparerStore{mirrorTailerStore: mirrorTailerStore{mirrorScriptStore: mirrorScriptStore{readFn: read}, tailFn: func(ctx context.Context, from Offset) iter.Seq2[*StoredEvent, error] { + return func(yield func(*StoredEvent, error) bool) { + if from == OffsetOldest { + for _, e := range batch { + if !yield(e, nil) { + return + } + } + } + <-ctx.Done() + } + }}, compareFn: compare} + } + replica := NewMemoryStore() + entered, release := make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + dst := &mirrorScriptStore{appendFn: func(ctx context.Context, e *Event) (Offset, error) { + if e.ID == "c" { + close(entered) + select { + case <-release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + return replica.Append(ctx, e) + }} + r := newTestReplicator(t, replicationConfig(src, dst)) + startReplicator(t, r) + lifecycleReceive(t, entered) + if p, _ := r.Confirmed(); p.Offset != OffsetOldest { + t.Fatalf("unfinished chunk confirmed: %+v", p) + } + target := replicationPoint(t, r, "02") + unblock() + if err := r.Wait(lifecycleContext(t), target); err != nil { + t.Fatal(err) + } + if mirrorCount(t, replica) != 3 { + t.Fatal("missing chunk suffix") + } + }) + } +} + +func TestReplicationDoesNotAcknowledgeHistoryReadErrors(t *testing.T) { + var fail atomic.Bool + fail.Store(true) + readFailed := make(chan struct{}, 1) + src := &mirrorComparerStore{mirrorScriptStore: mirrorScriptStore{readFn: func(_ context.Context, from Offset, _ int) ([]*StoredEvent, Offset, error) { + if from == OffsetNewest { + return nil, "01", nil + } + if fail.Load() { + select { + case readFailed <- struct{}{}: + default: + } + return nil, from, errors.New("required history is gone") + } + if from == OffsetOldest { + return []*StoredEvent{{Offset: "01", ID: "one"}}, "01", nil + } + return nil, from, nil + }}, compareFn: func(l, r Offset) (int, error) { return strings.Compare(string(l), string(r)), nil }} + r := newTestReplicator(t, replicationConfig(src, NewMemoryStore())) + startReplicator(t, r) + lifecycleReceive(t, readFailed) + if p, _ := r.Confirmed(); p.Offset != OffsetOldest { + t.Fatal("history gap was acknowledged") + } + fail.Store(false) + if err := r.Wait(lifecycleContext(t), replicationPoint(t, r, "01")); err != nil { + t.Fatal(err) + } +} + +func TestReplicationPositionAndCaptureErrors(t *testing.T) { + failure := errors.New("source unavailable") + src := &mirrorComparerStore{compareFn: func(l, r Offset) (int, error) { + if l == "invalid" { + return 0, failure + } + if l == "negative" { + return -1, nil + } + return strings.Compare(string(l), string(r)), nil + }} + r := newTestReplicator(t, replicationConfig(src, NewMemoryStore())) + for _, o := range []Offset{"invalid", "negative"} { + if _, err := r.Position(o); !errors.Is(err, ErrReplicationPosition) { + t.Fatalf("invalid position: %v", err) + } + } + for _, kind := range []string{"read error", "cancel during read", "historical events", "symbolic tail"} { + t.Run(kind, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + src.readFn = func(context.Context, Offset, int) ([]*StoredEvent, Offset, error) { + switch kind { + case "read error": + return nil, "", failure + case "cancel during read": + cancel() + return nil, "01", nil + case "historical events": + return []*StoredEvent{{}}, "01", nil + default: + return nil, OffsetNewest, nil + } + } + if _, err := r.Capture(ctx); err == nil { + t.Fatal("Capture accepted a failed/invalid tail") + } + }) + } +} + +func TestReplicationComparisonFailuresStopConfirmation(t *testing.T) { + failure := errors.New("cannot order these offsets") + for _, kind := range []string{"invalid checkpoint", "checkpoint vs tail", "wait target", "advance"} { + t.Run(kind, func(t *testing.T) { + var waitComparisons atomic.Int32 + src := &mirrorComparerStore{mirrorScriptStore: mirrorScriptStore{readFn: func(_ context.Context, from Offset, _ int) ([]*StoredEvent, Offset, error) { + if from == OffsetNewest { + return nil, "02", nil + } + if kind == "advance" { + return []*StoredEvent{{ID: "one", Offset: "invalid"}}, "invalid", nil + } + return nil, from, nil + }}, compareFn: func(l, r Offset) (int, error) { + if l == "invalid" || kind == "checkpoint vs tail" && l == "01" && r == "02" || kind == "wait target" && l == OffsetOldest && r == "02" && waitComparisons.Add(1) > 1 { + return 0, failure + } + return strings.Compare(string(l), string(r)), nil + }} + cfg := replicationConfig(src, NewMemoryStore()) + offsets := newMirrorLegacyOffsets() + cfg.Checkpoints = offsets + r := newTestReplicator(t, cfg) + if kind == "invalid checkpoint" { + offsets.offsets[r.checkpointID] = "invalid" + } + if kind == "checkpoint vs tail" { + offsets.offsets[r.checkpointID] = "01" + } + if kind == "wait target" { + startReplicator(t, r) + mirrorWaitFor(t, func() bool { _, ok := r.Confirmed(); return ok }) + if err := r.Wait(lifecycleContext(t), replicationPoint(t, r, "02")); !errors.Is(err, failure) { + t.Fatalf("Wait: %v", err) + } + } else { + if err := r.Run(lifecycleContext(t)); !errors.Is(err, failure) { + t.Fatalf("Run: %v", err) + } + } + }) + } +} + +func TestReplicationCancelWhileCheckpointIsUnavailable(t *testing.T) { + src, dst := NewMemoryStore(), NewMemoryStore() + cfg := replicationConfig(src, dst) + offsets := newMirrorLegacyOffsets() + cfg.Checkpoints = offsets + failed := make(chan struct{}, 1) + r := newTestReplicator(t, cfg, MirrorOnError(func(error) { + select { + case failed <- struct{}{}: + default: + } + })) + first := appendReplicationEvent(t, src, "first") + appendReplicationEvent(t, dst, "first") + offsets.offsets[r.checkpointID] = first + offsets.saveErr = errors.New("checkpoint outage") + target := replicationPoint(t, r, appendReplicationEvent(t, src, "second")) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- r.Run(ctx) }() + lifecycleReceive(t, failed) + cancel() + if err := lifecycleReceive(t, done); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if err := r.Wait(context.Background(), target); !errors.Is(err, ErrReplicationStopped) { + t.Fatalf("uncheckpointed target: %v", err) + } + // The second record already reached the destination. Restart may redeliver it, + // but cannot skip it based on the failed checkpoint save. + offsets.mu.Lock() + offsets.saveErr = nil + offsets.mu.Unlock() + resumed := newTestReplicator(t, cfg) + startReplicator(t, resumed) + if err := resumed.Wait(lifecycleContext(t), target); err != nil { + t.Fatal(err) + } + if mirrorCount(t, dst) != 3 { + t.Fatal("expected safe redelivery after checkpoint failure") + } +} + +type replicationWaitContext struct { + context.Context + entered chan struct{} + once sync.Once +} + +func (c *replicationWaitContext) Done() <-chan struct{} { + c.once.Do(func() { close(c.entered) }) + return c.Context.Done() +} +func TestReplicationWaitDeadlineDoesNotNeedARunner(t *testing.T) { + r := newTestReplicator(t, replicationConfig(NewMemoryStore(), NewMemoryStore())) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + observed := &replicationWaitContext{Context: ctx, entered: make(chan struct{})} + done := make(chan error, 1) + go func() { done <- r.Wait(observed, replicationPoint(t, r, OffsetOldest)) }() + lifecycleReceive(t, observed.entered) + cancel() + if err := lifecycleReceive(t, done); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } +} diff --git a/scripts/replication-smoke.go b/scripts/replication-smoke.go new file mode 100644 index 0000000..050af86 --- /dev/null +++ b/scripts/replication-smoke.go @@ -0,0 +1,213 @@ +//go:build ignore + +// Run from the repository workspace: go run ./scripts/replication-smoke.go +// A writer process acknowledges a replication barrier, is killed, and loses its +// entire local directory. Recovery reads only the reopened destination database. +package main + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + eventbus "github.com/jilio/ebu" + "github.com/jilio/ebu/stores/sqlite" +) + +const criticalRevision = 201 + +type remoteStore struct { + url string + client *http.Client +} + +func (s remoteStore) Append(ctx context.Context, e *eventbus.Event) (eventbus.Offset, error) { + data, err := json.Marshal(e) + if err != nil { + return "", err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.url, strings.NewReader(string(data))) + if err != nil { + return "", err + } + response, err := s.client.Do(req) + if err != nil { + return "", err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return "", fmt.Errorf("append: HTTP %d", response.StatusCode) + } + var offset eventbus.Offset + err = json.NewDecoder(response.Body).Decode(&offset) + return offset, err +} +func (remoteStore) Read(context.Context, eventbus.Offset, int) ([]*eventbus.StoredEvent, eventbus.Offset, error) { + return nil, "", fmt.Errorf("write-only destination") +} + +func writer(dir, url string) error { + local, err := sqlite.New(filepath.Join(dir, "source.db")) + if err != nil { + return err + } + defer local.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r, err := eventbus.NewReplicator(eventbus.ReplicatorConfig{Source: local, Destination: remoteStore{url, &http.Client{Timeout: 5 * time.Second}}, Checkpoints: local, ID: "smoke", Generation: "g1"}, eventbus.MirrorPollInterval(time.Millisecond)) + if err != nil { + return err + } + go func() { + if err := r.Run(ctx); err != nil { + fmt.Fprintln(os.Stderr, err) + } + }() + for i := 1; i <= criticalRevision+100; i++ { + data, _ := json.Marshal(map[string]int{"revision": i}) + offset, err := local.Append(ctx, &eventbus.Event{ID: fmt.Sprint(i), Type: "revision", Data: data}) + if err != nil { + return err + } + if i == criticalRevision { + p, err := r.Position(offset) + if err != nil { + return err + } + wait, stop := context.WithTimeout(ctx, 15*time.Second) + err = r.Wait(wait, p) + stop() + if err != nil { + return err + } + fmt.Println("ACK") + } + } + <-ctx.Done() + return ctx.Err() +} + +func smoke() error { + root, err := os.MkdirTemp("", "ebu-replication-smoke-") + if err != nil { + return err + } + defer os.RemoveAll(root) + source := filepath.Join(root, "worker") + if err := os.Mkdir(source, 0700); err != nil { + return err + } + destination := filepath.Join(root, "replica.db") + replica, err := sqlite.New(destination) + if err != nil { + return err + } + defer replica.Close() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + var e eventbus.Event + if err := json.NewDecoder(req.Body).Decode(&e); err != nil { + http.Error(w, err.Error(), 400) + return + } + offset, err := replica.Append(req.Context(), &e) + if err != nil { + http.Error(w, err.Error(), 500) + return + } + _ = json.NewEncoder(w).Encode(offset) + })) + defer server.Close() + executable, err := os.Executable() + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, executable, "writer", source, server.URL) + cmd.Stderr = os.Stderr + stdout, err := cmd.StdoutPipe() + if err != nil { + return err + } + if err := cmd.Start(); err != nil { + return err + } + waited := false + defer func() { + _ = cmd.Process.Kill() + if !waited { + _ = cmd.Wait() + } + }() + ack := make(chan bool, 1) + go func() { scanner := bufio.NewScanner(stdout); ack <- scanner.Scan() && scanner.Text() == "ACK" }() + select { + case ok := <-ack: + if !ok { + return fmt.Errorf("writer exited before confirmation") + } + case <-ctx.Done(): + return ctx.Err() + } + if err := cmd.Process.Kill(); err != nil { + return err + } + _ = cmd.Wait() + waited = true + server.Close() // Join any destination append already accepted before the kill. + if err := os.RemoveAll(source); err != nil { + return err + } + if err := replica.Close(); err != nil { + return err + } + recovered, err := sqlite.New(destination) + if err != nil { + return err + } + defer recovered.Close() + events, _, err := recovered.Read(context.Background(), eventbus.OffsetOldest, 0) + if err != nil { + return err + } + seen := make(map[string]bool) + for _, e := range events { + var record struct { + Revision int `json:"revision"` + } + if err := json.Unmarshal(e.Data, &record); err != nil { + return err + } + if e.Type != "revision" || fmt.Sprint(record.Revision) != e.ID { + return fmt.Errorf("corrupt replicated revision %q", e.ID) + } + seen[e.ID] = true + } + for i := 1; i <= criticalRevision; i++ { + if !seen[fmt.Sprint(i)] { + return fmt.Errorf("confirmed prefix lost revision %d", i) + } + } + fmt.Printf("PASS: killed source process, removed its files, recovered all %d confirmed revisions from reopened replica\n", criticalRevision) + return nil +} +func main() { + var err error + if len(os.Args) == 4 && os.Args[1] == "writer" { + err = writer(os.Args[2], os.Args[3]) + } else { + err = smoke() + } + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/stores/durablestream/options.go b/stores/durablestream/options.go index c93a3d6..47efee1 100644 --- a/stores/durablestream/options.go +++ b/stores/durablestream/options.go @@ -26,6 +26,7 @@ type config struct { retryAttempts int retryBaseDelay time.Duration decodeErrorHandler DecodeErrorHandler + strictDecoding bool } func defaultConfig() *config { @@ -110,3 +111,12 @@ func WithDecodeErrorHandler(handler DecodeErrorHandler) Option { c.decodeErrorHandler = handler } } + +// WithStrictDecoding makes Read and Tail fail on an undecodable +// event instead of skipping it. Use this for replication sources: skipping a +// corrupt event would let a prefix barrier acknowledge incomplete history. +// The default remains best-effort decoding for compatibility. Returned errors +// include the event index; WithDecodeErrorHandler only observes skipped events. +func WithStrictDecoding() Option { + return func(c *config) { c.strictDecoding = true } +} diff --git a/stores/durablestream/store.go b/stores/durablestream/store.go index 7fe4ad8..fd00b23 100644 --- a/stores/durablestream/store.go +++ b/stores/durablestream/store.go @@ -433,6 +433,9 @@ func (*Store) CompareOffsets(left, right eventbus.Offset) (int, error) { func (s *Store) decodeChunk(data []byte, next, chunkStart eventbus.Offset) (events []*eventbus.StoredEvent, allEmbedded bool, err error) { // Parse JSON array response (an empty body yields no events). var rawEvents []json.RawMessage + if s.cfg.strictDecoding && strings.TrimSpace(string(data)) == "null" { + return nil, false, fmt.Errorf("durablestream: decode chunk: null array") + } if len(data) > 0 { if err := json.Unmarshal(data, &rawEvents); err != nil { return nil, false, fmt.Errorf("durablestream: unmarshal response: %w", err) @@ -444,9 +447,15 @@ func (s *Store) decodeChunk(data []byte, next, chunkStart eventbus.Offset) (even allEmbedded = true lastRaw := len(rawEvents) - 1 for i, raw := range rawEvents { + if s.cfg.strictDecoding && strings.TrimSpace(string(raw)) == "null" { + return nil, false, fmt.Errorf("durablestream: decode event at index %d: null envelope", i) + } // Try to parse as event with embedded offset first var eventWithOffset storedEventWithOffset if err := json.Unmarshal(raw, &eventWithOffset); err != nil { + if s.cfg.strictDecoding { + return nil, false, fmt.Errorf("durablestream: decode event at index %d: %w", i, err) + } s.handleDecodeError(i, raw, err) continue } diff --git a/stores/durablestream/strict_decode_test.go b/stores/durablestream/strict_decode_test.go new file mode 100644 index 0000000..ea5bf41 --- /dev/null +++ b/stores/durablestream/strict_decode_test.go @@ -0,0 +1,80 @@ +package durablestream_test + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + eventbus "github.com/jilio/ebu" + ds "github.com/jilio/ebu/stores/durablestream" +) + +func TestStrictDecodingDoesNotSkipPoisonRecords(t *testing.T) { + for _, poison := range []string{`"not an object"`, `null`, `{"type":42}`} { + for _, mode := range []string{"read", "tail"} { + t.Run(mode+poison, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut { + w.WriteHeader(http.StatusCreated) + return + } + w.Header().Set("Stream-Next-Offset", "03") + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Stream-Up-To-Date", "true") + _, _ = w.Write([]byte(`[{"id":"a","type":"valid","data":{}},` + poison + `,{"id":"c","type":"valid","data":{}}]`)) + })) + defer srv.Close() + store, err := ds.New(srv.URL, "strict", ds.WithStrictDecoding(), ds.WithDecodeErrorHandler(func(error, []byte) { t.Error("strict mode skipped an event") })) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + var decodeErr error + switch mode { + case "read": + events, next, err := store.Read(ctx, eventbus.OffsetOldest, 0) + decodeErr = err + if len(events) != 0 || next != eventbus.OffsetOldest { + t.Fatal("read advanced past a malformed record") + } + case "tail": + for e, err := range store.Tail(ctx, eventbus.OffsetOldest) { + if e != nil { + t.Error("partial chunk escaped strict decoding") + } + decodeErr = err + break + } + } + if decodeErr == nil || !strings.Contains(decodeErr.Error(), "index 1") { + t.Fatalf("expected indexed decoding error, got %v", decodeErr) + } + }) + } + } +} + +func TestStrictDecodingRejectsNullChunk(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut { + w.WriteHeader(http.StatusCreated) + return + } + w.Header().Set("Stream-Next-Offset", "01") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`null`)) + })) + defer srv.Close() + store, err := ds.New(srv.URL, "strict", ds.WithStrictDecoding()) + if err != nil { + t.Fatal(err) + } + events, offset, err := store.Read(context.Background(), eventbus.OffsetOldest, 0) + if err == nil || len(events) != 0 || offset != eventbus.OffsetOldest { + t.Fatalf("null chunk advanced: %v %q %v", events, offset, err) + } +} From 14c2e95bd9103a50e2c83c533ff922f6560bf947 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sat, 5 Sep 2026 19:11:09 +1000 Subject: [PATCH 2/3] fix: confirm cursor-only progress and wake idle replication waits --- CHANGELOG.md | 3 ++- docs/REPLICATION.md | 11 +++++++++-- mirror.go | 23 +++++++++++++++++++++-- persist.go | 5 +++++ replication.go | 18 +++++++++++++----- replication_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 93 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 81cd90b..c51301e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. @@ -21,7 +23,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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 diff --git a/docs/REPLICATION.md b/docs/REPLICATION.md index a8aa4f4..a4fac18 100644 --- a/docs/REPLICATION.md +++ b/docs/REPLICATION.md @@ -67,6 +67,12 @@ configuration to resume. `MirrorPollInterval`, `MirrorDedupWindow`, `MirrorOnForward` and `MirrorOnError` also configure a Replicator. Observers execute on the copying goroutine; never wait for that replicator from an observer. +Replicator uses batched `Read`, including for sources with `Tail`: Read exposes +cursor-only advances across empty chunks, which Tail's event iterator cannot +report. `MirrorPollInterval` sets the idle polling period. A waiting operation +wakes an idle copier immediately; it does not bypass retry backoff after errors. +Plain Mirror continues to prefer Tail when available. + ## Mix ordinary and important writes Both modes use the same local append. Important operations add a wait: @@ -95,14 +101,15 @@ func appendWithReplication( ``` The application chooses its default and can override it per operation without -restarting the copier. Assign a stable event ID before attempting the local write +restarting the copier. Assign an event ID unique to that logical write before attempting the local write and make retries of the write idempotent. If the append succeeded but the wait timed out, keep the returned position and retry **the wait**, not the write. An append error can also be ambiguous; that case requires the source store's idempotency/reconciliation mechanism. `Position` checks the offset using the source comparer; it cannot prove that an -arbitrary token was actually issued by this log. Pass only a source Append result +arbitrary token was actually issued by this log. Concrete boundaries must name +stable prefixes: later appends cannot reuse a tail token with a larger meaning. Pass only a source Append result whose boundary covers that append, or use `Capture`. Do not use the offset on an arbitrary read event as evidence that the event was copied: a non-last member of a chunk can carry the chunk's start token. The copier confirms the end only after diff --git a/mirror.go b/mirror.go index 7a6183a..fb8d594 100644 --- a/mirror.go +++ b/mirror.go @@ -272,7 +272,7 @@ func mirrorRun(ctx context.Context, src, dst EventStore, subscriptionID string, return err } - if tailer, ok := src.(EventStoreTailer); ok { + if tailer, ok := src.(EventStoreTailer); ok && cfg.progress == nil { return m.runTail(ctx, tailer) } return m.runPoll(ctx) @@ -372,7 +372,7 @@ func (m *mirror) runPoll(ctx context.Context) error { if err := m.maybeReconcile(ctx); err != nil { return err } - if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil { + if err := m.waitForPoll(ctx); err != nil { return err } continue @@ -387,6 +387,25 @@ func (m *mirror) runPoll(ctx context.Context) error { } } +// A Replicator uses Read even for tail-capable stores: Tail cannot expose an +// empty chunk's advanced cursor, so it cannot confirm every captured boundary. +// An explicit waiter wakes idle polling without bypassing failure backoff. +func (m *mirror) waitForPoll(ctx context.Context) error { + if m.cfg.progress == nil { + return sleepCtx(ctx, m.cfg.pollInterval) + } + timer := time.NewTimer(m.cfg.pollInterval) + defer timer.Stop() + select { + case <-m.cfg.progress.wake: + return nil + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // runTail consumes a pushing source, restarting the tail after errors. func (m *mirror) runTail(ctx context.Context, tailer EventStoreTailer) error { for { diff --git a/persist.go b/persist.go index e8f261b..1ecf3e0 100644 --- a/persist.go +++ b/persist.go @@ -96,6 +96,11 @@ type EventStore interface { // either a StoredEvent.Offset or nextOffset. Resumable subscriptions then stop // immediately after the one unit that crossed their captured tail instead of // chasing concurrent appends. +// Concrete boundaries must identify stable prefixes within one log generation: +// appending more history must not change the meaning of an already issued tail +// token. Several read events can share an earlier token for safe redelivery; +// that does not make that token evidence of their inclusion in the prefix. +// CompareOffsets must be safe to call concurrently. type EventStoreOffsetComparer interface { CompareOffsets(left, right Offset) (int, error) } diff --git a/replication.go b/replication.go index 51e91ba..01e1853 100644 --- a/replication.go +++ b/replication.go @@ -40,8 +40,8 @@ type ReplicationPosition struct { // from opaque offsets, especially after a rebuilt log has caught up again. // // Source must implement EventStoreOffsetComparer, expose every event without -// filtering/skipping decode failures, and retain all history not yet replicated. A fresh generation starts at OffsetOldest and requires the -// full source history. Source and destination must be different logs with no +// filtering/skipping decode failures, and retain all history not yet replicated. +// A fresh generation starts at OffsetOldest and requires the full source history. Source and destination must be different logs with no // cyclic mirror topology. Destination must preserve all acknowledged records (or // a recovery-equivalent snapshot). There must be one owner of the checkpoint // namespace across processes; Replicator does not acquire distributed leases. @@ -71,6 +71,7 @@ type Replicator struct { checkpointID string mu sync.Mutex changed chan struct{} + wake chan struct{} started bool stopped bool runErr error @@ -109,14 +110,15 @@ func NewReplicator(config ReplicatorConfig, opts ...MirrorOption) (*Replicator, // aliasing another relationship's checkpoints. Plain Mirror IDs stay separate. key, _ := json.Marshal([2]string{config.ID, config.Generation}) r := &Replicator{config: config, comparer: comparer, mirrorConfig: cfg, - checkpointID: "ebu:replication:" + string(key), changed: make(chan struct{})} + checkpointID: "ebu:replication:" + string(key), changed: make(chan struct{}), wake: make(chan struct{}, 1)} cfg.progress = r return r, nil } // Position scopes an offset returned by Source.Append to this relationship. // It delegates token comparisons to the store; this is not an existence or -// provenance check. Pass only offsets actually issued by this source generation. OffsetNewest is rejected; use Capture to resolve it. +// provenance check. Pass only offsets actually issued by this source generation. +// OffsetNewest is rejected; use Capture to resolve it. func (r *Replicator) Position(offset Offset) (ReplicationPosition, error) { if offset == OffsetNewest { return ReplicationPosition{}, fmt.Errorf("%w: newest is symbolic", ErrReplicationPosition) @@ -153,7 +155,9 @@ func (r *Replicator) Capture(ctx context.Context) (ReplicationPosition, error) { } // Run copies until canceled or a fatal startup/checkpoint error. Transient -// mirror errors are retried in place. Concurrent/repeated Run calls fail with +// mirror errors are retried in place. Run uses batched Read so cursor-only +// advances (empty chunks) are visible; Wait wakes idle polling without bypassing +// retry backoff. Concurrent/repeated Run calls fail with // ErrReplicatorStarted; use a new instance to resume after this call returns. // Observers run on this goroutine: never call Wait on this replicator from one. func (r *Replicator) Run(ctx context.Context) (err error) { @@ -221,6 +225,10 @@ func (r *Replicator) Wait(ctx context.Context, target ReplicationPosition) error return runErr } select { + case r.wake <- struct{}{}: + default: + } + select { case <-changed: case <-ctx.Done(): return ctx.Err() diff --git a/replication_test.go b/replication_test.go index cd9993e..de52a9a 100644 --- a/replication_test.go +++ b/replication_test.go @@ -638,3 +638,46 @@ func TestReplicationWaitDeadlineDoesNotNeedARunner(t *testing.T) { t.Fatal(err) } } + +func TestReplicationConfirmsEmptyResumeBoundaryFromTailCapableSource(t *testing.T) { + // Tail has no way to yield a cursor-only update. A confirmed copier must use + // Read's nextOffset even for a source that also provides a pushing iterator. + source := &mirrorTailerComparerStore{mirrorTailerStore: mirrorTailerStore{ + mirrorScriptStore: mirrorScriptStore{readFn: func(_ context.Context, from Offset, _ int) ([]*StoredEvent, Offset, error) { return nil, "01", nil }}, + tailFn: func(ctx context.Context, _ Offset) iter.Seq2[*StoredEvent, error] { + return func(func(*StoredEvent, error) bool) { <-ctx.Done() } + }, + }, compareFn: func(l, r Offset) (int, error) { return strings.Compare(string(l), string(r)), nil }} + r := newTestReplicator(t, replicationConfig(source, NewMemoryStore())) + point, err := r.Capture(context.Background()) + if err != nil { + t.Fatal(err) + } + startReplicator(t, r) + ctx := lifecycleContext(t) + if err := r.Wait(ctx, point); err != nil { + t.Fatalf("empty boundary was never confirmed: %v", err) + } +} + +func TestReplicationWaitWakesIdlePolling(t *testing.T) { + memory := NewMemoryStore() + idle := make(chan struct{}) + var once sync.Once + source := &mirrorComparerStore{mirrorScriptStore: mirrorScriptStore{readFn: func(ctx context.Context, from Offset, n int) ([]*StoredEvent, Offset, error) { + events, next, err := memory.Read(ctx, from, n) + if from != OffsetNewest && len(events) == 0 { + once.Do(func() { close(idle) }) + } + return events, next, err + }}, compareFn: memory.CompareOffsets} + r := newTestReplicator(t, replicationConfig(source, NewMemoryStore()), MirrorPollInterval(time.Hour)) + startReplicator(t, r) + lifecycleReceive(t, idle) + target := replicationPoint(t, r, appendReplicationEvent(t, memory, "wake")) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := r.Wait(ctx, target); err != nil { + t.Fatalf("waiter did not wake idle copier: %v", err) + } +} From cb1aaa8c15d2be8571bb6baf495c6770a24b5903 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sat, 5 Sep 2026 19:12:12 +1000 Subject: [PATCH 3/3] fix: reject ambiguous non-UTF-8 replication identities --- docs/REPLICATION.md | 1 + replication.go | 7 ++++++- replication_test.go | 6 +++++- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/docs/REPLICATION.md b/docs/REPLICATION.md index a4fac18..d138bbf 100644 --- a/docs/REPLICATION.md +++ b/docs/REPLICATION.md @@ -56,6 +56,7 @@ go func() { done <- r.Run(runCtx) }() `EventStoreOffsetComparer`. The bundled memory, SQLite and Durable Streams stores do. The checkpoint store must accept the source's tokens verbatim: SQLite source checkpoints can live in SQLite; opaque remote tokens cannot necessarily do so. +ID and Generation must be valid UTF-8 strings. The tuple `(ID, Generation)` gets its own checkpoint namespace, separate from ordinary `Mirror` IDs. Constructing an object does not import an older plain Mirror checkpoint: a new relationship copies from `OffsetOldest`. diff --git a/replication.go b/replication.go index 01e1853..3bec1f3 100644 --- a/replication.go +++ b/replication.go @@ -7,6 +7,7 @@ import ( "fmt" "sync" "time" + "unicode/utf8" ) var ( @@ -33,7 +34,8 @@ type ReplicationPosition struct { Offset Offset `json:"offset"` } -// ReplicatorConfig binds one replication relationship. ID must identify the +// ReplicatorConfig binds one replication relationship. ID and Generation must +// be valid UTF-8 strings for unambiguous persisted/serialized identity. ID identifies the // source/destination pair. Generation is an application-owned immutable epoch: // change it after replacing, rebuilding or restoring EITHER log. Epoch changes // require a new Replicator and checkpoint namespace; they cannot be inferred @@ -90,6 +92,9 @@ func NewReplicator(config ReplicatorConfig, opts ...MirrorOption) (*Replicator, if config.ID == "" || config.Generation == "" { return nil, fmt.Errorf("eventbus: replication ID and generation are required") } + if !utf8.ValidString(config.ID) || !utf8.ValidString(config.Generation) { + return nil, fmt.Errorf("eventbus: replication ID and generation must be valid UTF-8") + } comparer, ok := config.Source.(EventStoreOffsetComparer) if !ok { return nil, fmt.Errorf("eventbus: replication source must implement EventStoreOffsetComparer") diff --git a/replication_test.go b/replication_test.go index de52a9a..4c90d55 100644 --- a/replication_test.go +++ b/replication_test.go @@ -57,7 +57,7 @@ func replicationPoint(t *testing.T, r *Replicator, o Offset) ReplicationPosition } func TestReplicatorValidation(t *testing.T) { src, dst := NewMemoryStore(), NewMemoryStore() - for _, name := range []string{"source", "destination", "checkpoints", "id", "generation", "comparer", "nil option", "bad option", "rewind"} { + for _, name := range []string{"source", "destination", "checkpoints", "id", "generation", "invalid ID", "invalid generation", "comparer", "nil option", "bad option", "rewind"} { t.Run(name, func(t *testing.T) { c := replicationConfig(src, dst) var opts []MirrorOption @@ -72,6 +72,10 @@ func TestReplicatorValidation(t *testing.T) { c.ID = "" case "generation": c.Generation = "" + case "invalid ID": + c.ID = string([]byte{0xff}) + case "invalid generation": + c.Generation = string([]byte{0xfe}) case "comparer": c.Source = &mirrorScriptStore{} case "nil option":