diff --git a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs
index 5d184f4a..6376cabd 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 00000000..f54de6f9
--- /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;
+ }
+ }
+}