Skip to content

fix(persistence): recognize streams present in either tier #623

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nmummau wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

fix(persistence): recognize streams present in either tier #623

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 17 additions & 5 deletions src/Core/src/Eventuous.Persistence/EventStore/TieredEventStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ namespace Eventuous;
/// <summary>
/// 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.
/// </summary>
/// <param name="hotStore"></param>
Expand All @@ -30,11 +30,23 @@ CancellationToken cancellationToken
public Task<AppendEventsResult[]> AppendEvents(IReadOnlyCollection<NewStreamAppend> appends, CancellationToken cancellationToken)
=> hotStore.AppendEvents(appends, cancellationToken);

/// <summary>
/// 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.
/// </summary>
public async Task<bool> 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(
Expand Down
170 changes: 170 additions & 0 deletions src/Core/test/Eventuous.Tests.Application/TieredEventStoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<Exception>(() => store.StreamExists(Stream, cancellation.Token));

if (cancelled) thrown.ShouldBeAssignableTo<OperationCanceledException>();
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<AppendEventsResult> Seed(IEventStore store) => store.AppendEvents(Stream, ExpectedStreamVersion.NoStream, Events(0, 1, 2), default);
static async Task<int[]> 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<StreamEvent> ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken)
=> inner.ReadEvents(stream, start, count, cancellationToken);
public IAsyncEnumerable<StreamEvent> 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<StreamEvent> ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) {
Token = cancellationToken;
try { await Task.FromException(failure); }
finally { Disposed = true; }
yield break;
}

public IAsyncEnumerable<StreamEvent> 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<StreamEvent> IEventReader.ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken)
=> Yield(events.Where(x => x.Revision >= start.Value).Take(count), cancellationToken);
IAsyncEnumerable<StreamEvent> 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<StreamEvent> Yield(IEnumerable<StreamEvent> events, [EnumeratorCancellation] CancellationToken ct) {
foreach (var evt in events) {
ct.ThrowIfCancellationRequested();
yield return evt;
}
await Task.CompletedTask;
}
}
}
Loading