From c4a047b77b5f92e320e7f67c18224e1bcd6afb94 Mon Sep 17 00:00:00 2001 From: Neal Mummau Date: Tue, 6 Oct 2026 00:10:54 -0400 Subject: [PATCH] fix(persistence): recognize streams present in either tier Use either tier for existence checks and probe read-only archives according to the IEventReader missing-stream contract. Preserve failures and cancellation; an empty existing stream still exists. Exercise bounded reads across tiers and hot-only writes, truncation, and deletion. Five regressions failed before the fix. All 36 application tests now pass on net10.0; TieredEventStore line coverage is 19/20. Fixes Eventuous/eventuous#609 --- .../EventStore/TieredEventStore.cs | 22 ++- .../TieredEventStoreTests.cs | 170 ++++++++++++++++++ 2 files changed, 187 insertions(+), 5 deletions(-) create mode 100644 src/Core/test/Eventuous.Tests.Application/TieredEventStoreTests.cs diff --git a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs index 5d184f4a5..6376cabd5 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs @@ -6,7 +6,7 @@ namespace Eventuous; /// /// Event store that appends events to the hot store, /// but reads from both hot store (recent events) and archive store (events missing from the hot store). -/// Truncation and deletion are only performed on hot store. Stream existence checks are performed on both stores. +/// Truncation and deletion are only performed on hot store. A stream exists if either tier contains it. /// It doesn't perform the archive itself, you need to use a connector to move events between hot and archive stores. /// /// @@ -30,11 +30,23 @@ CancellationToken cancellationToken public Task AppendEvents(IReadOnlyCollection appends, CancellationToken cancellationToken) => hotStore.AppendEvents(appends, cancellationToken); + /// + /// A stream exists if either tier contains it. For a read-only archive, a successful read + /// (including an empty existing stream) establishes existence; only StreamNotFound means absence. + /// public async Task StreamExists(StreamName stream, CancellationToken cancellationToken = default) { - var hotExists = await hotStore.StreamExists(stream, cancellationToken); - var archiveExists = archiveReader is IEventStore store && await store.StreamExists(stream, cancellationToken); - - return hotExists && archiveExists; + if (await hotStore.StreamExists(stream, cancellationToken).NoContext()) return true; + if (archiveReader is IEventStore store) return await store.StreamExists(stream, cancellationToken).NoContext(); + + try { + await using var events = archiveReader.ReadEvents(stream, StreamReadPosition.Start, 1, cancellationToken) + .GetAsyncEnumerator(cancellationToken); + await events.MoveNextAsync().NoContext(); + return true; + } + catch (StreamNotFound) { + return false; + } } public Task TruncateStream( diff --git a/src/Core/test/Eventuous.Tests.Application/TieredEventStoreTests.cs b/src/Core/test/Eventuous.Tests.Application/TieredEventStoreTests.cs new file mode 100644 index 000000000..f54de6f98 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Application/TieredEventStoreTests.cs @@ -0,0 +1,170 @@ +using System.Runtime.CompilerServices; +using Eventuous.Testing; +using Shouldly; + +namespace Eventuous.Tests.Application; + +public class TieredEventStoreTests { + static readonly StreamName Stream = new("tiered-test"); + + [Test] + [Arguments(false, false)] + [Arguments(true, false)] + [Arguments(false, true)] + [Arguments(true, true)] + public async Task Stream_exists_when_either_tier_contains_it(bool inHot, bool inArchive) { + var hot = new InMemoryEventStore(); + var archive = new InMemoryEventStore(); + if (inHot) await Seed(hot); + if (inArchive) await Seed(archive); + + (await new TieredEventStore(hot, archive).StreamExists(Stream)).ShouldBe(inHot || inArchive); + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task Existence_can_be_determined_from_a_read_only_archive(bool present) { + var archive = new InMemoryEventStore(); + if (present) await Seed(archive); + var store = new TieredEventStore(new InMemoryEventStore(), new ReaderOnly(archive)); + (await store.StreamExists(Stream)).ShouldBe(present); + } + + [Test] + public async Task Empty_existing_archive_stream_still_exists() { + var archive = new InMemoryEventStore(); + await Seed(archive); + await archive.TruncateStream(Stream, new(2), ExpectedStreamVersion.Any, default); + var store = new TieredEventStore(new InMemoryEventStore(), new ReaderOnly(archive)); + (await store.StreamExists(Stream)).ShouldBeTrue(); + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task Archive_read_failure_is_not_reported_as_a_missing_stream(bool cancelled) { + using var cancellation = new CancellationTokenSource(); + if (cancelled) cancellation.Cancel(); + Exception failure = cancelled ? new OperationCanceledException(cancellation.Token) : new IOException("Archive unavailable"); + var archive = new FailingReader(failure); + var store = new TieredEventStore(new InMemoryEventStore(), archive); + + var thrown = await Should.ThrowAsync(() => store.StreamExists(Stream, cancellation.Token)); + + if (cancelled) thrown.ShouldBeAssignableTo(); + else thrown.ShouldBeSameAs(failure); + archive.Token.ShouldBe(cancellation.Token); + archive.Disposed.ShouldBeTrue(); + } + + [Test] + public async Task Hot_stream_exists_even_when_archive_is_unavailable() { + var hot = new InMemoryEventStore(); + await Seed(hot); + var archive = new FailingReader(new IOException("Archive unavailable")); + + (await new TieredEventStore(hot, archive).StreamExists(Stream)).ShouldBeTrue(); + archive.Disposed.ShouldBeFalse(); + } + + [Test] + public async Task Reads_span_tiers_in_both_directions_without_duplicates() { + var events = Enumerable.Range(0, 6) + .Select(i => new StreamEvent(Guid.NewGuid(), i, new(), "application/json", i, DateTime.UtcNow)).ToArray(); + // The archive overlaps the hot window at revision 3. + var store = new TieredEventStore(new ReadWindow(events[3..]), new ReadWindow(events[..4])); + var forward = await store.ReadEvents(Stream, new(1), 4, true, default); + var backward = await store.ReadEventsBackwards(Stream, new(5), 4, true, default); + forward.Select(x => x.Revision).ShouldBe(new long[] { 1, 2, 3, 4 }); + backward.Select(x => x.Revision).ShouldBe(new long[] { 5, 4, 3, 2 }); + forward.Select(x => x.Payload).ShouldBe(new object[] { 1, 2, 3, 4 }); + backward.Select(x => x.Payload).ShouldBe(new object[] { 5, 4, 3, 2 }); + forward.Select(x => x.FromArchive).ShouldBe(new[] { true, true, false, false }); + backward.Select(x => x.FromArchive).ShouldBe(new[] { false, false, false, true }); + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task Appends_only_modify_the_hot_store(bool batch) { + var hot = new InMemoryEventStore(); + var archive = new InMemoryEventStore(); + await Seed(archive); + var store = new TieredEventStore(hot, archive); + var events = Events(10, 11); + if (batch) { + var results = await store.AppendEvents(new[] { new NewStreamAppend(Stream, ExpectedStreamVersion.NoStream, events) }, default); + results.Single().NextExpectedVersion.ShouldBe(1); + } + else { + var result = await store.AppendEvents(Stream, ExpectedStreamVersion.NoStream, events, default); + result.NextExpectedVersion.ShouldBe(1); + } + (await Read(hot)).ShouldBe(new[] { 10, 11 }); + (await Read(archive)).ShouldBe(new[] { 0, 1, 2 }); + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task Destructive_operations_leave_archive_events_intact(bool delete) { + var hot = new InMemoryEventStore(); + var archive = new InMemoryEventStore(); + await Seed(hot); + await Seed(archive); + var store = new TieredEventStore(hot, archive); + if (delete) { + await store.DeleteStream(Stream, new(2)); + (await hot.StreamExists(Stream, default)).ShouldBeFalse(); + } + else { + await store.TruncateStream(Stream, new(0), new(2)); + (await Read(hot)).ShouldBe(new[] { 1, 2 }); + } + (await Read(archive)).ShouldBe(new[] { 0, 1, 2 }); + (await store.StreamExists(Stream)).ShouldBeTrue(); + } + + static NewStreamEvent[] Events(params int[] values) => values.Select(x => new NewStreamEvent(Guid.NewGuid(), x, new())).ToArray(); + static Task Seed(IEventStore store) => store.AppendEvents(Stream, ExpectedStreamVersion.NoStream, Events(0, 1, 2), default); + static async Task Read(IEventReader reader) => (await reader.ReadEvents(Stream, StreamReadPosition.Start, 10, true, default)).Select(x => (int)x.Payload!).ToArray(); + + sealed class ReaderOnly(IEventReader inner) : IEventReader { + public IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) + => inner.ReadEvents(stream, start, count, cancellationToken); + public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) + => inner.ReadEventsBackwards(stream, start, count, cancellationToken); + } + + sealed class FailingReader(Exception failure) : IEventReader { + public CancellationToken Token { get; private set; } + public bool Disposed { get; private set; } + + public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { + Token = cancellationToken; + try { await Task.FromException(failure); } + finally { Disposed = true; } + yield break; + } + + public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) + => throw new NotSupportedException(); + } + + // Controlled read windows retain original revisions after truncation, just like provider stores. + sealed class ReadWindow(StreamEvent[] events) : InMemoryEventStore, IEventReader { + IAsyncEnumerable IEventReader.ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) + => Yield(events.Where(x => x.Revision >= start.Value).Take(count), cancellationToken); + IAsyncEnumerable IEventReader.ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) + => Yield(events.Where(x => x.Revision <= start.Value).Reverse().Take(count), cancellationToken); + + static async IAsyncEnumerable Yield(IEnumerable events, [EnumeratorCancellation] CancellationToken ct) { + foreach (var evt in events) { + ct.ThrowIfCancellationRequested(); + yield return evt; + } + await Task.CompletedTask; + } + } +}