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..c51301e 100644 --- a/CHANGELOG.md +++ b/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 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..d138bbf --- /dev/null +++ b/docs/REPLICATION.md @@ -0,0 +1,220 @@ +# 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. +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`. + +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. + +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: + +```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 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. 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 +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..fb8d594 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 @@ -261,7 +272,7 @@ func Mirror(ctx context.Context, src, dst EventStore, subscriptionID string, off 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) @@ -361,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 @@ -376,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 { @@ -469,26 +499,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..1ecf3e0 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. @@ -92,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 new file mode 100644 index 0000000..3bec1f3 --- /dev/null +++ b/replication.go @@ -0,0 +1,271 @@ +package eventbus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + "unicode/utf8" +) + +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 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 +// 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{} + wake 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") + } + 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") + } + 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{}), 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. +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. 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) { + 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 r.wake <- struct{}{}: + default: + } + 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..4c90d55 --- /dev/null +++ b/replication_test.go @@ -0,0 +1,687 @@ +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", "invalid ID", "invalid 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 "invalid ID": + c.ID = string([]byte{0xff}) + case "invalid generation": + c.Generation = string([]byte{0xfe}) + 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) + } +} + +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) + } +} 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) + } +}