From 12c0020b878677a7b3b9be3eac93033aa227c6e9 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 12 Aug 2026 18:47:52 +0200 Subject: [PATCH 01/12] fix(persistence): make reads stream and add paged read-to-end KurrentDB ReadEvents/ReadEventsBackwards materialized the entire requested range (raw ResolvedEvent[] plus deserialized StreamEvent[]) before the first yield, so IAsyncEnumerable consumers got O(stream) memory instead of streaming. Rewrite both as true streaming iterators that map exceptions per enumerator advance and hold at most one deserialized event at a time. Add IEventReader.ReadStreamToEnd extension that reads a stream to the end in pages, so count: int.MaxValue stops being the read-to-end idiom, and make ReadStream delegate to it, fixing page advancement for truncated streams. Document the memory semantics on IEventReader. New contract tests exposed two pre-existing provider bugs, also fixed: - Sqlite reads never threw StreamNotFound for a missing stream; empty read results are now verified with StreamExists in SqlEventStoreBase - Postgres and SqlServer overflowed reading backwards from StreamReadPosition.End (long.MaxValue into an INT parameter); the client parameter is now clamped to the 32-bit position range Closes #567 Co-Authored-By: Claude Fable 5 --- .../EventStore/IEventReader.cs | 6 ++ .../EventStore/StoreFunctions.cs | 72 +++++++++++--- .../Store/Read.cs | 96 +++++++++++++++++++ .../KurrentDBEventStore.cs | 87 +++++++++-------- .../Store/StreamingReadTests.cs | 71 ++++++++++++++ .../src/Eventuous.Postgresql/PostgresStore.cs | 3 +- .../Eventuous.Sql.Base/SqlEventStoreBase.cs | 6 ++ .../src/Eventuous.SqlServer/SqlServerStore.cs | 3 +- 8 files changed, 287 insertions(+), 57 deletions(-) create mode 100644 src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs diff --git a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs index 8268287b9..72290ed2c 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs @@ -7,6 +7,10 @@ public interface IEventReader { /// /// Read a fixed number of events from an existing stream as an async enumerable. /// Throws if the stream does not exist. + /// Implementations either stream events as they arrive from the store, or buffer up to + /// events before yielding, so memory usage can grow with . To read a whole stream, + /// use , which reads in pages, instead of passing + /// as the count. /// /// Stream name /// Where to start reading events @@ -18,6 +22,8 @@ public interface IEventReader { /// /// Read a number of events from a given stream, backwards (from the stream end). /// Throws if the stream does not exist. + /// Implementations either stream events as they arrive from the store, or buffer up to + /// events before yielding, so memory usage can grow with . /// /// Stream name /// Where to start reading events diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index ca2ee79f2..7a09b6b3d 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -1,6 +1,8 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +using System.Runtime.CompilerServices; + namespace Eventuous; public static class StoreFunctions { @@ -148,6 +150,59 @@ CancellationToken cancellationToken } } + /// + /// Reads a stream from the given position to the end, as an async enumerable. + /// Events are read in pages of and yielded as they arrive, so the whole stream + /// is never buffered in memory. Use this instead of calling + /// with as the count. + /// + /// Name of the stream to read from + /// Stream position to start reading from + /// Number of events to read per page. It caps the amount of events a buffering + /// implementation of holds in memory at a time. + /// Set to false to complete without yielding anything when the stream isn't found, + /// instead of throwing . Default is true. + /// Cancellation token + /// An async enumerable of events retrieved from the stream + public async IAsyncEnumerable ReadStreamToEnd( + StreamName streamName, + StreamReadPosition start, + int pageSize = 500, + bool failIfNotFound = true, + [EnumeratorCancellation] CancellationToken cancellationToken = default + ) { + var position = start; + + while (true) { + var yielded = 0; + long lastRevision = 0; + + await using var enumerator = eventReader.ReadEvents(streamName, position, pageSize, cancellationToken).GetAsyncEnumerator(cancellationToken); + + while (true) { + bool moved; + + try { + moved = await enumerator.MoveNextAsync().NoContext(); + } catch (StreamNotFound) when (!failIfNotFound) { + yield break; + } + + if (!moved) break; + + var evt = enumerator.Current; + yielded++; + lastRevision = evt.Revision; + + yield return evt; + } + + if (yielded < pageSize) yield break; + + position = new(lastRevision + 1); + } + } + /// /// Reads a stream from the event store to a collection of /// @@ -163,23 +218,10 @@ public async Task ReadStream( bool failIfNotFound = true, CancellationToken cancellationToken = default ) { - const int pageSize = 500; - var streamEvents = new List(); - var position = start; - - try { - while (true) { - var events = await eventReader.ReadEvents(streamName, position, pageSize, failIfNotFound, cancellationToken).NoContext(); - streamEvents.AddRange(events); - - if (events.Length < pageSize) break; - - position = new(position.Value + events.Length); - } - } catch (StreamNotFound) when (!failIfNotFound) { - return []; + await foreach (var evt in eventReader.ReadStreamToEnd(streamName, start, failIfNotFound: failIfNotFound, cancellationToken: cancellationToken).NoContext(cancellationToken)) { + streamEvents.Add(evt); } return [.. streamEvents]; diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs index 25adbbfba..d8f7b0c62 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs @@ -146,6 +146,102 @@ public async Task ShouldReturnWhenReadingBackwards(CancellationToken cancellatio await Assert.That(result.Length).IsEqualTo(5); } + [Test] + [Category("Store")] + public async Task ShouldThrowWhenReadingMissingStream(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + + await Assert.ThrowsAsync(() => _fixture.EventStore.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); + } + + [Test] + [Category("Store")] + public async Task ShouldThrowWhenReadingMissingStreamBackwards(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + + await Assert.ThrowsAsync(() => _fixture.EventStore.ReadEventsBackwards(streamName, StreamReadPosition.End, 10, true, cancellationToken)); + } + + [Test] + [Category("Store")] + public async Task ShouldReadStreamToEnd(CancellationToken cancellationToken) { + object[] events = [.. _fixture.CreateEvents(25)]; + var streamName = Helpers.GetStreamName(); + await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload)!; + await Assert.That(actual).IsEquivalentTo(events); + } + + [Test] + [Category("Store")] + public async Task ShouldReadStreamToEndWithExactPageMultiple(CancellationToken cancellationToken) { + object[] events = [.. _fixture.CreateEvents(20)]; + var streamName = Helpers.GetStreamName(); + await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload)!; + await Assert.That(actual).IsEquivalentTo(events); + } + + [Test] + [Category("Store")] + public async Task ShouldReadStreamToEndFromPosition(CancellationToken cancellationToken) { + object[] events = [.. _fixture.CreateEvents(25)]; + var streamName = Helpers.GetStreamName(); + await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, new(10), pageSize: 10, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + var expected = events.Skip(10); + var actual = result.Select(x => x.Payload!); + await Assert.That(actual).IsEquivalentTo(expected); + } + + [Test] + [Category("Store")] + public async Task ShouldThrowWhenReadingMissingStreamToEnd(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + + await Assert.ThrowsAsync(ReadFunc); + + return; + + async Task ReadFunc() { + await foreach (var _ in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, cancellationToken: cancellationToken)) { } + } + } + + [Test] + [Category("Store")] + public async Task ShouldReturnNothingWhenReadingMissingStreamToEnd(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, failIfNotFound: false, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + await Assert.That(result).IsEmpty(); + } + [Test] [Category("Store")] public async Task ShouldThrowWhenReadingBackwardsFromNegativePosition(CancellationToken cancellationToken) { diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs index 165bac57d..30ab02114 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs @@ -216,48 +216,63 @@ EventData ToEventData(NewStreamEvent streamEvent) { } /// - public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - var read = _client.ReadStreamAsync(Direction.Forwards, stream, start.AsStreamPosition(), count, cancellationToken: cancellationToken); - - var events = await TryExecute( - async () => { - var resolvedEvents = await read.ToArrayAsync(cancellationToken).NoContext(); - - return ToStreamEvents(resolvedEvents); - }, + public IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken = default) + => EnumerateStream( + () => _client.ReadStreamAsync(Direction.Forwards, stream, start.AsStreamPosition(), count, cancellationToken: cancellationToken), stream, - true, () => new("Unable to read {Count} starting at {Start} events from {Stream}", count, start, stream), - (s, ex) => new ReadFromStreamException(s, ex) + cancellationToken ); - foreach (var evt in events) yield return evt; - } - /// - public async IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - var read = _client.ReadStreamAsync( - Direction.Backwards, + public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken = default) + => EnumerateStream( + () => _client.ReadStreamAsync(Direction.Backwards, stream, start.AsStreamPosition(), count, resolveLinkTos: true, cancellationToken: cancellationToken), stream, - start.AsStreamPosition(), - count, - resolveLinkTos: true, - cancellationToken: cancellationToken + () => new("Unable to read {Count} events backwards from {Stream}", count, stream), + cancellationToken ); - var events = await TryExecute( - async () => { - var resolvedEvents = await read.ToArrayAsync(cancellationToken).NoContext(); + // Events are yielded as they arrive from the server, so a read holds at most one + // deserialized event at a time, regardless of the requested count. + // The exception mapping wraps each advance of the source enumerator instead of the whole + // loop because iterators can't yield from inside a try block with a catch clause. + async IAsyncEnumerable EnumerateStream( + Func> read, + string stream, + Func getError, + [EnumeratorCancellation] CancellationToken cancellationToken + ) { + await using var enumerator = read().GetAsyncEnumerator(cancellationToken); - return ToStreamEvents(resolvedEvents); - }, - stream, - true, - () => new("Unable to read {Count} events backwards from {Stream}", count, stream), - (s, ex) => new ReadFromStreamException(s, ex) - ); + while (true) { + var moved = false; + StreamEvent? streamEvent = null; + + try { + moved = await enumerator.MoveNextAsync().NoContext(); - foreach (var evt in events) yield return evt; + if (moved) streamEvent = ToStreamEvent(enumerator.Current); + } catch (StreamNotFoundException) { + LogStreamStreamNotFound(stream); + + throw new StreamNotFound(stream); + } catch (OperationCanceledException) { + throw; + } catch (Exception ex) { + var (message, args) = getError(); + // ReSharper disable once TemplateIsNotCompileTimeConstantProblem +#pragma warning disable CA2254 + _logger.LogWarning(ex, message, args); +#pragma warning restore CA2254 + + throw new ReadFromStreamException(stream, ex); + } + + if (!moved) yield break; + + if (streamEvent != null) yield return streamEvent.Value; + } } /// @@ -362,14 +377,6 @@ StreamEvent AsStreamEvent(object payload) ); } - StreamEvent[] ToStreamEvents(ResolvedEvent[] resolvedEvents) - => [ - .. resolvedEvents - .Select(ToStreamEvent) - .Where(x => x != null) - .Select(x => x!.Value) - ]; - record ErrorInfo(string Message, params object[] Args); [LoggerMessage(LogLevel.Warning, "Stream {stream} not found")] diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs new file mode 100644 index 000000000..bd66c21e6 --- /dev/null +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs @@ -0,0 +1,71 @@ +using Eventuous.KurrentDB; +using Eventuous.Sut.Domain; +using Eventuous.Tests.Persistence.Base.Fixtures; + +namespace Eventuous.Tests.KurrentDB.Store; + +[ClassDataSource] +public class StreamingReadTests { + readonly StoreFixture _fixture; + + public StreamingReadTests(StoreFixture fixture) { + fixture.TypeMapper.RegisterKnownEventTypes(typeof(BookingEvents.BookingImported).Assembly); + _fixture = fixture; + } + + const int EventCount = 100; + + [Test] + [Category("Store")] + public async Task ShouldStreamEventsForwardsWithoutBufferingWholeRead(CancellationToken cancellationToken) { + var serializer = new CountingSerializer(_fixture.Serializer); + var store = new KurrentDBEventStore(_fixture.Client, serializer); + + object[] events = [.. _fixture.CreateEvents(EventCount)]; + var streamName = Helpers.GetStreamName(); + await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); + + var deserializedAtFirstYield = 0; + + await foreach (var _ in store.ReadEvents(streamName, StreamReadPosition.Start, EventCount, cancellationToken)) { + if (deserializedAtFirstYield == 0) deserializedAtFirstYield = serializer.DeserializedCount; + } + + await Assert.That(deserializedAtFirstYield).IsEqualTo(1); + await Assert.That(serializer.DeserializedCount).IsEqualTo(EventCount); + } + + [Test] + [Category("Store")] + public async Task ShouldStreamEventsBackwardsWithoutBufferingWholeRead(CancellationToken cancellationToken) { + var serializer = new CountingSerializer(_fixture.Serializer); + var store = new KurrentDBEventStore(_fixture.Client, serializer); + + object[] events = [.. _fixture.CreateEvents(EventCount)]; + var streamName = Helpers.GetStreamName(); + await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream); + + var deserializedAtFirstYield = 0; + + await foreach (var _ in store.ReadEventsBackwards(streamName, new(EventCount - 1), EventCount, cancellationToken)) { + if (deserializedAtFirstYield == 0) deserializedAtFirstYield = serializer.DeserializedCount; + } + + await Assert.That(deserializedAtFirstYield).IsEqualTo(1); + await Assert.That(serializer.DeserializedCount).IsEqualTo(EventCount); + } + + class CountingSerializer(IEventSerializer inner) : IEventSerializer { + int _deserializedCount; + + public int DeserializedCount => _deserializedCount; + + public DeserializationResult DeserializeEvent(ReadOnlySpan data, string eventType, string contentType) { + Interlocked.Increment(ref _deserializedCount); + + return inner.DeserializeEvent(data, eventType, contentType); + } + + public SerializationResult SerializeEvent(object evt) => inner.SerializeEvent(evt); + } +} diff --git a/src/Postgres/src/Eventuous.Postgresql/PostgresStore.cs b/src/Postgres/src/Eventuous.Postgresql/PostgresStore.cs index 4017811c7..25bf8b1db 100644 --- a/src/Postgres/src/Eventuous.Postgresql/PostgresStore.cs +++ b/src/Postgres/src/Eventuous.Postgresql/PostgresStore.cs @@ -56,7 +56,8 @@ protected override DbCommand GetReadCommand(NpgsqlConnection connection, StreamN protected override DbCommand GetReadBackwardsCommand(NpgsqlConnection connection, StreamName stream, StreamReadPosition start, int count) => connection.GetCommand(Schema.ReadStreamBackwards) .Add("_stream_name", NpgsqlDbType.Varchar, stream.ToString()) - .Add("_from_position", NpgsqlDbType.Integer, start.Value) + // Stream positions are 32-bit, so StreamReadPosition.End gets clamped, and the function trims it to the stream head + .Add("_from_position", NpgsqlDbType.Integer, (int)Math.Min(start.Value, int.MaxValue)) .Add("_count", NpgsqlDbType.Integer, count); protected override bool IsStreamNotFound(Exception exception) diff --git a/src/Relational/src/Eventuous.Sql.Base/SqlEventStoreBase.cs b/src/Relational/src/Eventuous.Sql.Base/SqlEventStoreBase.cs index 8f2a41d21..8729106d7 100644 --- a/src/Relational/src/Eventuous.Sql.Base/SqlEventStoreBase.cs +++ b/src/Relational/src/Eventuous.Sql.Base/SqlEventStoreBase.cs @@ -103,6 +103,9 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR var events = await ReadInternal(stream, start, count, cancellationToken).NoContext(); + // A plain query can't tell a missing stream from a read past the stream end + if (events.Length == 0 && !await StreamExists(stream, cancellationToken).NoContext()) throw new StreamNotFound(stream); + foreach (var evt in events) yield return evt; } @@ -112,6 +115,9 @@ public async IAsyncEnumerable ReadEventsBackwards(StreamName stream var events = await ReadInternalBackwards(stream, start, count, cancellationToken).NoContext(); + // A plain query can't tell a missing stream from a read past the stream end + if (events.Length == 0 && !await StreamExists(stream, cancellationToken).NoContext()) throw new StreamNotFound(stream); + foreach (var evt in events) yield return evt; } diff --git a/src/SqlServer/src/Eventuous.SqlServer/SqlServerStore.cs b/src/SqlServer/src/Eventuous.SqlServer/SqlServerStore.cs index 118953802..fd5ee7484 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/SqlServerStore.cs +++ b/src/SqlServer/src/Eventuous.SqlServer/SqlServerStore.cs @@ -41,7 +41,8 @@ protected override DbCommand GetReadBackwardsCommand(SqlConnection connection, S => connection .GetStoredProcCommand(Schema.ReadStreamBackwards) .Add("@stream_name", SqlDbType.NVarChar, stream.ToString()) - .Add("@from_position", SqlDbType.Int, start.Value) + // Stream positions are 32-bit, so StreamReadPosition.End gets clamped, and the procedure trims it to the stream head + .Add("@from_position", SqlDbType.Int, (int)Math.Min(start.Value, int.MaxValue)) .Add("@count", SqlDbType.Int, count); protected override bool IsStreamNotFound(Exception exception) => exception is SqlException e && e.Message.StartsWith("StreamNotFound"); From 76ea4454160a2b2f5f30933b4483d2f54311327c Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 14:37:41 +0200 Subject: [PATCH 02/12] fix(persistence): harden paged reads per review findings Address review findings on the streaming reads change: - KurrentDBEventStore reads now deliver the requested count even when non-deserializable system events are skipped, issuing follow-up reads from the last received position. A short read now reliably means the stream end, which ReadStreamToEnd's paging termination depends on. - ReadStreamToEnd rejects non-positive page sizes instead of spinning forever on readers that complete immediately for count <= 0. - TieredEventReader no longer throws StreamNotFound when reading past the end of an existing stream; it throws only when both tiers report the stream missing. - RedisStore distinguishes a missing stream from a read past the stream end by checking key existence when a read returns nothing. - IEventReader docs now state the short-read and past-end contract. Co-Authored-By: Claude Fable 5 --- .../EventStore/IEventReader.cs | 2 + .../EventStore/StoreFunctions.cs | 85 ++++++++++------- .../EventStore/TieredEventReader.cs | 51 ++++++---- .../Store/Read.cs | 15 +++ .../Store/TieredStoreTests.cs | 38 ++++++++ .../KurrentDBEventStore.cs | 93 +++++++++++++------ .../Store/StreamingReadTests.cs | 68 ++++++++++++++ .../Store/TieredStoreTests.cs | 10 ++ src/Redis/src/Eventuous.Redis/RedisStore.cs | 11 ++- .../test/Eventuous.Tests.Redis/Store/Read.cs | 18 ++++ 10 files changed, 308 insertions(+), 83 deletions(-) diff --git a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs index 72290ed2c..612dbe82f 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs @@ -11,6 +11,8 @@ public interface IEventReader { /// events before yielding, so memory usage can grow with . To read a whole stream, /// use , which reads in pages, instead of passing /// as the count. + /// Implementations must yield exactly events unless the end of the stream is reached, + /// and must return an empty sequence, not throw, when reading past the end of an existing stream. /// /// Stream name /// Where to start reading events diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index 7a09b6b3d..3705c9b77 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -164,43 +164,16 @@ CancellationToken cancellationToken /// instead of throwing . Default is true. /// Cancellation token /// An async enumerable of events retrieved from the stream - public async IAsyncEnumerable ReadStreamToEnd( - StreamName streamName, - StreamReadPosition start, - int pageSize = 500, - bool failIfNotFound = true, - [EnumeratorCancellation] CancellationToken cancellationToken = default + public IAsyncEnumerable ReadStreamToEnd( + StreamName streamName, + StreamReadPosition start, + int pageSize = 500, + bool failIfNotFound = true, + CancellationToken cancellationToken = default ) { - var position = start; - - while (true) { - var yielded = 0; - long lastRevision = 0; - - await using var enumerator = eventReader.ReadEvents(streamName, position, pageSize, cancellationToken).GetAsyncEnumerator(cancellationToken); - - while (true) { - bool moved; - - try { - moved = await enumerator.MoveNextAsync().NoContext(); - } catch (StreamNotFound) when (!failIfNotFound) { - yield break; - } + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(pageSize); - if (!moved) break; - - var evt = enumerator.Current; - yielded++; - lastRevision = evt.Revision; - - yield return evt; - } - - if (yielded < pageSize) yield break; - - position = new(lastRevision + 1); - } + return ReadToEnd(eventReader, streamName, start, pageSize, failIfNotFound, cancellationToken); } /// @@ -227,4 +200,46 @@ public async Task ReadStream( return [.. streamEvents]; } } + + // Relies on readers yielding exactly `count` events unless the stream end is reached: + // a page shorter than pageSize means there is nothing left to read + static async IAsyncEnumerable ReadToEnd( + IEventReader eventReader, + StreamName streamName, + StreamReadPosition start, + int pageSize, + bool failIfNotFound, + [EnumeratorCancellation] CancellationToken cancellationToken + ) { + var position = start; + + while (true) { + var yielded = 0; + long lastRevision = 0; + + await using var enumerator = eventReader.ReadEvents(streamName, position, pageSize, cancellationToken).GetAsyncEnumerator(cancellationToken); + + while (true) { + bool moved; + + try { + moved = await enumerator.MoveNextAsync().NoContext(); + } catch (StreamNotFound) when (!failIfNotFound) { + yield break; + } + + if (!moved) break; + + var evt = enumerator.Current; + yielded++; + lastRevision = evt.Revision; + + yield return evt; + } + + if (yielded < pageSize) yield break; + + position = new(lastRevision + 1); + } + } } diff --git a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs index 3095489b8..4b7a9336d 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs @@ -13,14 +13,24 @@ namespace Eventuous; /// Event reader pointing to archive store public class TieredEventReader(IEventReader hotReader, IEventReader archiveReader) : IEventReader { public async IAsyncEnumerable ReadEvents(StreamName streamName, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { - var hotEvents = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken).NoContext(); + var (hotEvents, hotNotFound) = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken).NoContext(); - var archivedEvents = hotEvents.Length switch { - > 0 when hotEvents[0].Revision > start.Value - => (await LoadStreamEvents(archiveReader, streamName, start, (int)hotEvents[0].Revision, cancellationToken).NoContext()).Select(x => x with { FromArchive = true }), - 0 => (await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken).NoContext()).Select(x => x with { FromArchive = true }), - _ => [] - }; + IEnumerable archivedEvents; + var archiveNotFound = false; + + switch (hotEvents.Length) { + case > 0 when hotEvents[0].Revision > start.Value: { + (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, (int)hotEvents[0].Revision, cancellationToken).NoContext(); + archivedEvents = events.Select(x => x with { FromArchive = true }); + + break; + } + case 0: + (var archived, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken).NoContext(); + archivedEvents = archived.Select(x => x with { FromArchive = true }); break; + default: + archivedEvents = []; break; + } var combined = archivedEvents.Concat(hotEvents).Distinct(Comparer); var any = false; @@ -31,28 +41,31 @@ public async IAsyncEnumerable ReadEvents(StreamName streamName, Str yield return evt; } - if (!any) throw new StreamNotFound(streamName); + // No events with both tiers reporting a missing stream means the stream doesn't exist; + // otherwise an empty result can mean the read window is past the stream end + if (!any && hotNotFound && archiveNotFound) throw new StreamNotFound(streamName); } public async IAsyncEnumerable ReadEventsBackwards(StreamName streamName, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { - var hotEvents = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken, backwards: true).NoContext(); + var (hotEvents, hotNotFound) = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken, backwards: true).NoContext(); IEnumerable archivedEvents; + var archiveNotFound = false; switch (hotEvents.Length) { case > 0 when hotEvents.Length < count: { // Hot store returned fewer events than requested, fill the gap from archive var lastHotRevision = hotEvents[^1].Revision; - archivedEvents = (await LoadStreamEvents(archiveReader, streamName, new(lastHotRevision - 1), count - hotEvents.Length, cancellationToken, backwards: true).NoContext()) - .Select(x => x with { FromArchive = true }); + (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, new(lastHotRevision - 1), count - hotEvents.Length, cancellationToken, backwards: true).NoContext(); + archivedEvents = events.Select(x => x with { FromArchive = true }); break; } case 0: // Hot store has no events, try archive for the full range - archivedEvents = (await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken, backwards: true).NoContext()) - .Select(x => x with { FromArchive = true }); break; + (var archived, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken, backwards: true).NoContext(); + archivedEvents = archived.Select(x => x with { FromArchive = true }); break; default: archivedEvents = []; break; } @@ -66,10 +79,12 @@ public async IAsyncEnumerable ReadEventsBackwards(StreamName stream yield return evt; } - if (!any) throw new StreamNotFound(streamName); + // No events with both tiers reporting a missing stream means the stream doesn't exist; + // otherwise an empty result can mean the read window is past the stream end + if (!any && hotNotFound && archiveNotFound) throw new StreamNotFound(streamName); } - static async Task LoadStreamEvents( + static async Task<(StreamEvent[] Events, bool NotFound)> LoadStreamEvents( IEventReader reader, StreamName streamName, StreamReadPosition startPosition, @@ -78,11 +93,13 @@ static async Task LoadStreamEvents( bool backwards = false ) { try { - return backwards + var events = backwards ? await reader.ReadEventsBackwards(streamName, startPosition, localCount, true, cancellationToken).NoContext() : await reader.ReadEvents(streamName, startPosition, localCount, true, cancellationToken).NoContext(); + + return (events, false); } catch (StreamNotFound) { - return []; + return ([], true); } } diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs index d8f7b0c62..fbec3133c 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs @@ -214,6 +214,21 @@ public async Task ShouldReadStreamToEndFromPosition(CancellationToken cancellati await Assert.That(actual).IsEquivalentTo(expected); } + [Test] + [Category("Store")] + public async Task ShouldRejectInvalidPageSizeReadingToEnd(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + + await Assert.ThrowsAsync(() => Read(0)); + await Assert.ThrowsAsync(() => Read(-1)); + + return; + + async Task Read(int pageSize) { + await foreach (var _ in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: pageSize, cancellationToken: cancellationToken)) { } + } + } + [Test] [Category("Store")] public async Task ShouldThrowWhenReadingMissingStreamToEnd(CancellationToken cancellationToken) { diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs index 9f7ca38f1..bf034c483 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs @@ -28,6 +28,44 @@ protected async Task Should_load_hot_and_archive() { await Assert.That(loaded.Skip(50).Select(x => x.FromArchive)).DoesNotContain(true); } + protected async Task Should_return_empty_reading_past_end() { + const int count = 10; + + var (tieredReader, stream, _) = await SeedTieredStream(count); + + var loaded = await tieredReader.ReadEvents(stream, new(count), 5, true, CancellationToken.None); + + await Assert.That(loaded).IsEmpty(); + } + + protected async Task Should_read_stream_to_end_with_exact_page_multiple() { + const int count = 100; + + var (tieredReader, stream, testEvents) = await SeedTieredStream(count); + + var loaded = new List(); + + // 100 events with page size 50 forces a final read past the stream end + await foreach (var evt in tieredReader.ReadStreamToEnd(stream, StreamReadPosition.Start, pageSize: 50)) { + loaded.Add(evt); + } + + var actual = loaded.Select(x => (TestEventForTiers)x.Payload!); + await Assert.That(actual).IsEquivalentTo(testEvents); + } + + async Task<(TieredEventReader Reader, StreamName Stream, TestEventForTiers[] Events)> SeedTieredStream(int count) { + var store = _storeFixture.EventStore; + var archive = new ArchiveStore(_storeFixture.EventStore); + var testEvents = TestEventForTiers.CreateMany(count).ToArray(); + var stream = new StreamName($"Test-{Guid.NewGuid():N}"); + + await store.Store(stream, ExpectedStreamVersion.NoStream, testEvents); + await archive.Store(stream, ExpectedStreamVersion.NoStream, testEvents); + + return (new(store, archive), stream, testEvents); + } + readonly StoreFixtureBase _storeFixture; protected TieredStoreTestsBase(StoreFixtureBase storeFixture) { diff --git a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs index 30ab02114..b62837830 100644 --- a/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs +++ b/src/KurrentDB/src/Eventuous.KurrentDB/KurrentDBEventStore.cs @@ -218,8 +218,10 @@ EventData ToEventData(NewStreamEvent streamEvent) { /// public IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken = default) => EnumerateStream( - () => _client.ReadStreamAsync(Direction.Forwards, stream, start.AsStreamPosition(), count, cancellationToken: cancellationToken), + (from, remaining) => _client.ReadStreamAsync(Direction.Forwards, stream, from ?? start.AsStreamPosition(), remaining, cancellationToken: cancellationToken), + forwards: true, stream, + count, () => new("Unable to read {Count} starting at {Start} events from {Stream}", count, start, stream), cancellationToken ); @@ -227,51 +229,86 @@ public IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPos /// public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken = default) => EnumerateStream( - () => _client.ReadStreamAsync(Direction.Backwards, stream, start.AsStreamPosition(), count, resolveLinkTos: true, cancellationToken: cancellationToken), + (from, remaining) => _client.ReadStreamAsync(Direction.Backwards, stream, from ?? start.AsStreamPosition(), remaining, resolveLinkTos: true, cancellationToken: cancellationToken), + forwards: false, stream, + count, () => new("Unable to read {Count} events backwards from {Stream}", count, stream), cancellationToken ); // Events are yielded as they arrive from the server, so a read holds at most one // deserialized event at a time, regardless of the requested count. + // Non-deserializable system events are skipped and compensated for with follow-up + // reads, so the enumeration delivers `count` events unless the stream end is reached — + // paged readers rely on a short read meaning the end of the stream. // The exception mapping wraps each advance of the source enumerator instead of the whole // loop because iterators can't yield from inside a try block with a catch clause. async IAsyncEnumerable EnumerateStream( - Func> read, - string stream, - Func getError, - [EnumeratorCancellation] CancellationToken cancellationToken + Func> read, + bool forwards, + string stream, + int count, + Func getError, + [EnumeratorCancellation] CancellationToken cancellationToken ) { - await using var enumerator = read().GetAsyncEnumerator(cancellationToken); + var remaining = count; + StreamPosition? from = null; + + while (remaining > 0) { + var requested = remaining; + var received = 0; + long lastRaw = 0; + + await using var enumerator = read(from, requested).GetAsyncEnumerator(cancellationToken); + + while (true) { + var moved = false; + StreamEvent? streamEvent = null; + + try { + moved = await enumerator.MoveNextAsync().NoContext(); + + if (moved) { + received++; + lastRaw = enumerator.Current.OriginalEventNumber.ToInt64(); + streamEvent = ToStreamEvent(enumerator.Current); + } + } catch (StreamNotFoundException) { + LogStreamStreamNotFound(stream); + + throw new StreamNotFound(stream); + } catch (OperationCanceledException) { + throw; + } catch (Exception ex) { + var (message, args) = getError(); + // ReSharper disable once TemplateIsNotCompileTimeConstantProblem +#pragma warning disable CA2254 + _logger.LogWarning(ex, message, args); +#pragma warning restore CA2254 - while (true) { - var moved = false; - StreamEvent? streamEvent = null; + throw new ReadFromStreamException(stream, ex); + } - try { - moved = await enumerator.MoveNextAsync().NoContext(); + if (!moved) break; - if (moved) streamEvent = ToStreamEvent(enumerator.Current); - } catch (StreamNotFoundException) { - LogStreamStreamNotFound(stream); + if (streamEvent != null) { + remaining--; - throw new StreamNotFound(stream); - } catch (OperationCanceledException) { - throw; - } catch (Exception ex) { - var (message, args) = getError(); - // ReSharper disable once TemplateIsNotCompileTimeConstantProblem -#pragma warning disable CA2254 - _logger.LogWarning(ex, message, args); -#pragma warning restore CA2254 - - throw new ReadFromStreamException(stream, ex); + yield return streamEvent.Value; + } } - if (!moved) yield break; + // Fewer events received than requested means the stream end was reached + if (received < requested) yield break; + + // Nothing was skipped and the requested count is delivered + if (remaining == 0) yield break; + + // Reading backwards can't continue past the first stream event + if (!forwards && lastRaw == 0) yield break; - if (streamEvent != null) yield return streamEvent.Value; + from = StreamPosition.FromInt64(forwards ? lastRaw + 1 : lastRaw - 1); } } diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs index bd66c21e6..e3abbede2 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/StreamingReadTests.cs @@ -1,6 +1,7 @@ using Eventuous.KurrentDB; using Eventuous.Sut.Domain; using Eventuous.Tests.Persistence.Base.Fixtures; +using KurrentDB.Client; namespace Eventuous.Tests.KurrentDB.Store; @@ -55,6 +56,73 @@ public async Task ShouldStreamEventsBackwardsWithoutBufferingWholeRead(Cancellat await Assert.That(serializer.DeserializedCount).IsEqualTo(EventCount); } + [Test] + [Category("Store")] + public async Task ShouldReadRequestedCountWhenSystemEventsAreSkipped(CancellationToken cancellationToken) { + var (streamName, events) = await SeedStreamWithSystemEvent(cancellationToken); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadEvents(streamName, StreamReadPosition.Start, events.Length, cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload)!; + await Assert.That(actual).IsEquivalentTo(events); + } + + [Test] + [Category("Store")] + public async Task ShouldReadRequestedCountBackwardsWhenSystemEventsAreSkipped(CancellationToken cancellationToken) { + var (streamName, events) = await SeedStreamWithSystemEvent(cancellationToken); + + var result = new List(); + + await foreach (var evt in _fixture.EventStore.ReadEventsBackwards(streamName, StreamReadPosition.End, events.Length, cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload).Reverse()!; + await Assert.That(actual).IsEquivalentTo(events); + } + + [Test] + [Category("Store")] + public async Task ShouldReadStreamToEndWhenSystemEventsAreSkipped(CancellationToken cancellationToken) { + var (streamName, events) = await SeedStreamWithSystemEvent(cancellationToken); + + var result = new List(); + + // The system event lands inside the first page, which then yields fewer events than the page size + await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 6, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload)!; + await Assert.That(actual).IsEquivalentTo(events); + } + + // Seeds a stream of 12 events where revision 5 is a non-deserializable $-typed event, + // which the store skips when reading. Returns the 11 deserializable events. + async Task<(StreamName Stream, object[] Events)> SeedStreamWithSystemEvent(CancellationToken cancellationToken) { + var streamName = Helpers.GetStreamName(); + object[] first = [.. _fixture.CreateEvents(5)]; + object[] rest = [.. _fixture.CreateEvents(6)]; + + await _fixture.AppendEvents(streamName, first, ExpectedStreamVersion.NoStream); + + await _fixture.Client.AppendToStreamAsync( + streamName.ToString(), + StreamState.Any, + [new EventData(Uuid.NewUuid(), "$test-skipped", "{}"u8.ToArray())], + cancellationToken: cancellationToken + ); + + await _fixture.AppendEvents(streamName, rest, ExpectedStreamVersion.Any); + + return (streamName, [.. first, .. rest]); + } + class CountingSerializer(IEventSerializer inner) : IEventSerializer { int _deserializedCount; diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs index e1efd488b..08c192b8d 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs @@ -9,4 +9,14 @@ public class TieredStoreTests(StoreFixture storeFixture) : TieredStoreTestsBase< public async Task Esdb_should_load_hot_and_archive() { await Should_load_hot_and_archive(); } + + [Test] + public async Task Esdb_should_return_empty_reading_past_end() { + await Should_return_empty_reading_past_end(); + } + + [Test] + public async Task Esdb_should_read_stream_to_end_with_exact_page_multiple() { + await Should_read_stream_to_end_with_exact_page_multiple(); + } } \ No newline at end of file diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index ab7edb3ab..778670e70 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -43,10 +43,15 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR var result = await _getDatabase().StreamReadAsync(stream.ToString(), start.Value.ToRedisValue(), count).NoContext(); if (result == null! || result.Length == 0) { - throw new StreamNotFound(stream); + // An empty result can also mean the read window is past the stream end + if (!await _getDatabase().KeyExistsAsync(stream.ToString()).NoContext()) { + throw new StreamNotFound(stream); + } + + events = []; + } else { + events = [.. result.Select(x => ToStreamEvent(x, _serializer, _metaSerializer))]; } - - events = [.. result.Select(x => ToStreamEvent(x, _serializer, _metaSerializer))]; } catch (InvalidOperationException e) when (e.Message.Contains("Reading is not allowed after reader was completed") || cancellationToken.IsCancellationRequested) { throw new OperationCanceledException("Redis read operation terminated", e, cancellationToken); diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 8a7649224..5e8041fdd 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -49,6 +49,24 @@ public async Task ShouldReadTail(CancellationToken cancellationToken) { await Assert.That(actual).IsEquivalentTo(events2); } + [Test] + public async Task ShouldReturnEmptyReadingPastEnd(CancellationToken cancellationToken) { + var events = CreateEvents(10).ToArray(); + var streamName = GetStreamName(); + var appended = await fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream, cancellationToken); + + var result = await fixture.EventReader.ReadEvents(streamName, new((long)appended.GlobalPosition + 1000), 10, true, cancellationToken); + + await Assert.That(result).IsEmpty(); + } + + [Test] + public async Task ShouldThrowWhenReadingMissingStream(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); + } + [Test] public async Task ShouldReadHead(CancellationToken cancellationToken) { // ReSharper disable once CoVariantArrayConversion From 90badd7f3c97fb36558f67ca7de60422ffc287b0 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 15:18:28 +0200 Subject: [PATCH 03/12] fix(persistence): bound tiered reads and make Redis reads inclusive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the second review round: - TieredEventReader bounds the archive gap request and the combined result to the requested count, so a read across a real archive/hot boundary no longer yields more events than asked for. Reading backwards past a hot store that bottoms out at revision 0 no longer crashes constructing a negative read position. - RedisStore reads use an inclusive range read (XRANGE) instead of the exclusive XREAD, matching the IEventReader position contract and the paged read extensions that advance from the last revision + 1 — pages no longer silently skip the event at the page boundary. Co-Authored-By: Claude Fable 5 --- .../EventStore/TieredEventReader.cs | 10 ++-- .../Store/TieredStoreTests.cs | 49 ++++++++++++++----- .../Store/TieredStoreTests.cs | 10 ++++ src/Redis/src/Eventuous.Redis/RedisStore.cs | 4 +- .../test/Eventuous.Tests.Redis/Store/Read.cs | 21 +++++++- 5 files changed, 78 insertions(+), 16 deletions(-) diff --git a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs index 4b7a9336d..08d6ffe3b 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs @@ -20,7 +20,10 @@ public async IAsyncEnumerable ReadEvents(StreamName streamName, Str switch (hotEvents.Length) { case > 0 when hotEvents[0].Revision > start.Value: { - (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, (int)hotEvents[0].Revision, cancellationToken).NoContext(); + // Fill the gap before the first hot event from the archive, bounded by the requested count + var gapCount = (int)Math.Min(count, hotEvents[0].Revision - start.Value); + + (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, gapCount, cancellationToken).NoContext(); archivedEvents = events.Select(x => x with { FromArchive = true }); break; @@ -32,7 +35,7 @@ public async IAsyncEnumerable ReadEvents(StreamName streamName, Str archivedEvents = []; break; } - var combined = archivedEvents.Concat(hotEvents).Distinct(Comparer); + var combined = archivedEvents.Concat(hotEvents).Distinct(Comparer).Take(count); var any = false; foreach (var evt in combined) { @@ -53,7 +56,8 @@ public async IAsyncEnumerable ReadEventsBackwards(StreamName stream var archiveNotFound = false; switch (hotEvents.Length) { - case > 0 when hotEvents.Length < count: { + // When the hot store read reached revision 0, no events can precede it + case > 0 when hotEvents.Length < count && hotEvents[^1].Revision > 0: { // Hot store returned fewer events than requested, fill the gap from archive var lastHotRevision = hotEvents[^1].Revision; diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs index bf034c483..8377a49cb 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/TieredStoreTests.cs @@ -9,17 +9,9 @@ public abstract class TieredStoreTestsBase where TContainer : Docker protected async Task Should_load_hot_and_archive() { const int count = 100; - var store = _storeFixture.EventStore; - var archive = new ArchiveStore(_storeFixture.EventStore); - var testEvents = TestEventForTiers.CreateMany(count).ToArray(); - var stream = new StreamName($"Test-{Guid.NewGuid():N}"); - - await store.Store(stream, ExpectedStreamVersion.NoStream, testEvents); - await archive.Store(stream, ExpectedStreamVersion.NoStream, testEvents); + var (combined, stream, testEvents) = await SeedTieredStream(count, truncateHotAt: 50); - await store.TruncateStream(stream, new(50), ExpectedStreamVersion.Any); - var combined = new TieredEventReader(store, archive); - var loaded = (await combined.ReadStream(stream, StreamReadPosition.Start)).ToArray(); + var loaded = (await combined.ReadStream(stream, StreamReadPosition.Start)).ToArray(); var actual = loaded.Select(x => (TestEventForTiers)x.Payload!); await Assert.That(actual).IsEquivalentTo(testEvents); @@ -28,6 +20,26 @@ protected async Task Should_load_hot_and_archive() { await Assert.That(loaded.Skip(50).Select(x => x.FromArchive)).DoesNotContain(true); } + protected async Task Should_read_bounded_count_across_tier_boundary() { + const int count = 100; + + var (combined, stream, testEvents) = await SeedTieredStream(count, truncateHotAt: 50); + + // The first 50 events only exist in the archive, the hot store starts at revision 50 + var firstPage = await combined.ReadEvents(stream, StreamReadPosition.Start, 50, true, CancellationToken.None); + + await Assert.That(firstPage.Length).IsEqualTo(50); + await Assert.That(firstPage.Select(x => (TestEventForTiers)x.Payload!)).IsEquivalentTo(testEvents.Take(50)); + + var loaded = new List(); + + await foreach (var evt in combined.ReadStreamToEnd(stream, StreamReadPosition.Start, pageSize: 50)) { + loaded.Add(evt); + } + + await Assert.That(loaded.Select(x => (TestEventForTiers)x.Payload!)).IsEquivalentTo(testEvents); + } + protected async Task Should_return_empty_reading_past_end() { const int count = 10; @@ -54,7 +66,18 @@ protected async Task Should_read_stream_to_end_with_exact_page_multiple() { await Assert.That(actual).IsEquivalentTo(testEvents); } - async Task<(TieredEventReader Reader, StreamName Stream, TestEventForTiers[] Events)> SeedTieredStream(int count) { + protected async Task Should_read_backwards_more_than_available() { + const int count = 10; + + var (combined, stream, testEvents) = await SeedTieredStream(count); + + // Requesting more events than the stream holds reads the hot store down to revision 0 + var loaded = await combined.ReadEventsBackwards(stream, StreamReadPosition.End, count * 2, true, CancellationToken.None); + + await Assert.That(loaded.Select(x => (TestEventForTiers)x.Payload!).Reverse()).IsEquivalentTo(testEvents); + } + + async Task<(TieredEventReader Reader, StreamName Stream, TestEventForTiers[] Events)> SeedTieredStream(int count, long? truncateHotAt = null) { var store = _storeFixture.EventStore; var archive = new ArchiveStore(_storeFixture.EventStore); var testEvents = TestEventForTiers.CreateMany(count).ToArray(); @@ -63,6 +86,10 @@ protected async Task Should_read_stream_to_end_with_exact_page_multiple() { await store.Store(stream, ExpectedStreamVersion.NoStream, testEvents); await archive.Store(stream, ExpectedStreamVersion.NoStream, testEvents); + if (truncateHotAt != null) { + await store.TruncateStream(stream, new(truncateHotAt.Value), ExpectedStreamVersion.Any); + } + return (new(store, archive), stream, testEvents); } diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs index 08c192b8d..08d8c4ec9 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/Store/TieredStoreTests.cs @@ -19,4 +19,14 @@ public async Task Esdb_should_return_empty_reading_past_end() { public async Task Esdb_should_read_stream_to_end_with_exact_page_multiple() { await Should_read_stream_to_end_with_exact_page_multiple(); } + + [Test] + public async Task Esdb_should_read_bounded_count_across_tier_boundary() { + await Should_read_bounded_count_across_tier_boundary(); + } + + [Test] + public async Task Esdb_should_read_backwards_more_than_available() { + await Should_read_backwards_more_than_available(); + } } \ No newline at end of file diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 778670e70..57db72ab7 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -40,7 +40,9 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR StreamEvent[] events; try { - var result = await _getDatabase().StreamReadAsync(stream.ToString(), start.Value.ToRedisValue(), count).NoContext(); + // Range read is inclusive of the start position, matching the IEventReader contract + // and the paged read extensions, which advance pages from the last revision + 1 + var result = await _getDatabase().StreamRangeAsync(stream.ToString(), start.Value.ToRedisValue(), count: count).NoContext(); if (result == null! || result.Length == 0) { // An empty result can also mean the read window is past the stream end diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 5e8041fdd..f3b7b9e50 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -43,12 +43,31 @@ public async Task ShouldReadTail(CancellationToken cancellationToken) { var events2 = CreateEvents(10).ToArray(); await fixture.AppendEvents(streamName, events2, ExpectedStreamVersion.Any, cancellationToken); - var result = await fixture.EventReader.ReadEvents(streamName, new((long)position), 100, true, cancellationToken); + // The read position is inclusive, so start from the position right after the first batch + var result = await fixture.EventReader.ReadEvents(streamName, new((long)position + 1), 100, true, cancellationToken); IEnumerable actual = result.Select(x => x.Payload)!; await Assert.That(actual).IsEquivalentTo(events2); } + [Test] + public async Task ShouldReadStreamToEndAcrossPages(CancellationToken cancellationToken) { + // Keep the batch small so the auto-generated Redis ID sequence numbers stay within + // the single digit that the position encoding can represent + var events = CreateEvents(8).ToArray(); + var streamName = GetStreamName(); + await fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream, cancellationToken); + + var result = new List(); + + await foreach (var evt in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 3, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + IEnumerable actual = result.Select(x => x.Payload)!; + await Assert.That(actual).IsEquivalentTo(events); + } + [Test] public async Task ShouldReturnEmptyReadingPastEnd(CancellationToken cancellationToken) { var events = CreateEvents(10).ToArray(); From 83747abf76ba4d01a94680cdcfda55a03eea09e2 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 15:35:32 +0200 Subject: [PATCH 04/12] fix(redis): make stream positions round-trip for store-written entries Address the third review round: - The append_events Redis function now assigns explicit entry IDs (millisecond-0, bumping the millisecond past the last entry when needed) instead of auto-generated ones, so every position the store writes round-trips through the millisecond*10+sequence encoding and paged reads no longer silently truncate same-millisecond bursts. - Reading a legacy entry whose auto-generated ID carries a sequence number above 9 now throws NotSupportedException with a clear message instead of silently garbling the position. - Relax IEventReader/ReadStreamToEnd memory docs to promise memory proportional to count/page size rather than capped at one page, matching the tiered reader which briefly holds up to two bounded pages. - Stabilize the Redis test fixture: abortConnect=false stops the first connection attempt from aborting when it races the freshly started container. Co-Authored-By: Claude Fable 5 --- .../EventStore/IEventReader.cs | 13 ++++---- .../EventStore/StoreFunctions.cs | 5 +-- src/Redis/src/Eventuous.Redis/RedisStore.cs | 2 +- .../Eventuous.Redis/Scripts/AppendEvents.lua | 31 ++++++++++++++++--- .../src/Eventuous.Redis/Tools/Conversions.cs | 12 +++++++ .../Fixtures/IntegrationFixture.cs | 4 ++- .../test/Eventuous.Tests.Redis/Store/Read.cs | 29 ++++++++++++++--- 7 files changed, 77 insertions(+), 19 deletions(-) diff --git a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs index 612dbe82f..6e69d92c0 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs @@ -7,10 +7,10 @@ public interface IEventReader { /// /// Read a fixed number of events from an existing stream as an async enumerable. /// Throws if the stream does not exist. - /// Implementations either stream events as they arrive from the store, or buffer up to - /// events before yielding, so memory usage can grow with . To read a whole stream, - /// use , which reads in pages, instead of passing - /// as the count. + /// Implementations either stream events as they arrive from the store, or buffer events in an amount + /// proportional to before yielding, so memory usage can grow with + /// . To read a whole stream, use , + /// which reads in pages, instead of passing as the count. /// Implementations must yield exactly events unless the end of the stream is reached, /// and must return an empty sequence, not throw, when reading past the end of an existing stream. /// @@ -24,8 +24,9 @@ public interface IEventReader { /// /// Read a number of events from a given stream, backwards (from the stream end). /// Throws if the stream does not exist. - /// Implementations either stream events as they arrive from the store, or buffer up to - /// events before yielding, so memory usage can grow with . + /// Implementations either stream events as they arrive from the store, or buffer events in an amount + /// proportional to before yielding, so memory usage can grow with + /// . /// /// Stream name /// Where to start reading events diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index 3705c9b77..aaa226078 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -158,8 +158,9 @@ CancellationToken cancellationToken /// /// Name of the stream to read from /// Stream position to start reading from - /// Number of events to read per page. It caps the amount of events a buffering - /// implementation of holds in memory at a time. + /// Number of events to read per page. It bounds the memory a buffering + /// implementation of uses: such implementations hold at most a small + /// multiple of a page in memory at a time (e.g. a tiered reader combining two stores). /// Set to false to complete without yielding anything when the stream isn't found, /// instead of throwing . Default is true. /// Cancellation token diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 57db72ab7..5cfb435b6 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -139,6 +139,6 @@ static StreamEvent ToStreamEvent(StreamEntry evt, IEventSerializer serializer, I }; StreamEvent AsStreamEvent(object payload) - => new(Guid.Parse(evt[MessageId].ToString()), payload, meta ?? new Metadata(), ContentType, evt.Id.ToLong(), DateTime.Parse(evt[Created]!, CultureInfo.InvariantCulture)); + => new(Guid.Parse(evt[MessageId].ToString()), payload, meta ?? new Metadata(), ContentType, evt.Id.ToRevision(), DateTime.Parse(evt[Created]!, CultureInfo.InvariantCulture)); } } diff --git a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua index 5a8831b47..80a2ab475 100644 --- a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua +++ b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua @@ -1,5 +1,17 @@ #!lua name=append_events +-- Entry IDs are assigned explicitly as '-0' with the millisecond part bumped past +-- the last entry when needed. Auto-generated IDs ('*') bump the sequence part instead, and the +-- client-side position encoding can only represent sequence numbers 0-9. +local function last_id_ms(key) + local entries = redis.call('XREVRANGE', key, '+', '-', 'COUNT', 1) + if #entries == 0 then + return 0 + end + local id = entries[1][1] + return tonumber(string.sub(id, 1, string.find(id, '-', 1, true) - 1)) +end + local function append_events(keys, args) local stream_name = keys[1] local expected_version = tonumber(keys[2]) @@ -22,20 +34,29 @@ local function append_events(keys, args) local global_position local items_inserted = 0 + local time = redis.call('TIME') + local now_ms = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000) + local stream_ms = last_id_ms(stream_name) + local all_ms = last_id_ms('_all') + for i=1, table.getn(args), 4 do + stream_ms = math.max(now_ms, stream_ms + 1) + local stream_position = redis.call( - 'XADD', stream_name, '*', + 'XADD', stream_name, string.format('%.0f', stream_ms) .. '-0', 'message_id', args[i], - 'message_type', args[i+1], - 'json_data', args[i+2], + 'message_type', args[i+1], + 'json_data', args[i+2], 'json_metadata', args[i+3], 'created', created ) + all_ms = math.max(now_ms, all_ms + 1) + global_position = redis.call( - 'XADD', '_all', '*', - 'stream', stream_name, + 'XADD', '_all', string.format('%.0f', all_ms) .. '-0', + 'stream', stream_name, 'position', stream_position ) diff --git a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs index cb0a882ed..d44a6d98b 100644 --- a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs +++ b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs @@ -9,6 +9,18 @@ public static long ToLong(this RedisValue value) { return long.Parse(first) * 10 + long.Parse(second); } + public static long ToRevision(this RedisValue value) { + var (first, second) = new Split(Ensure.NotNull(value).AsSpan()); + var sequence = long.Parse(second); + + return sequence <= 9 + ? long.Parse(first) * 10 + sequence + : throw new NotSupportedException( + $"Redis stream entry ID {value} can't be represented as a stream position: the position encoding only supports ID sequence numbers 0-9. " + + "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store." + ); + } + public static ulong ToULong(this ReadOnlySpan valueString) { var (first, second) = new Split(valueString); return ulong.Parse(first) * 10 + ulong.Parse(second); diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs index e237509d3..0c37883ba 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs @@ -37,7 +37,9 @@ public async Task InitializeAsync() { IDatabase GetDb() { // FLUSHDB in test teardown is an admin command; StackExchange.Redis 3.x enforces the // admin gate for raw commands issued through Execute as well. - var muxer = ConnectionMultiplexer.Connect($"{connString},allowAdmin=true"); + // abortConnect=false keeps the multiplexer retrying when the first connection attempt + // races the freshly started container. + var muxer = ConnectionMultiplexer.Connect($"{connString},allowAdmin=true,abortConnect=false"); return muxer.GetDatabase(); } diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index f3b7b9e50..d5f0d93e1 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -1,3 +1,4 @@ +using System.Globalization; using Eventuous.Tests.Redis.Fixtures; using Shouldly; using static Eventuous.Tests.Redis.Store.Helpers; @@ -52,15 +53,14 @@ public async Task ShouldReadTail(CancellationToken cancellationToken) { [Test] public async Task ShouldReadStreamToEndAcrossPages(CancellationToken cancellationToken) { - // Keep the batch small so the auto-generated Redis ID sequence numbers stay within - // the single digit that the position encoding can represent - var events = CreateEvents(8).ToArray(); + // A single batch this large lands in one millisecond, so all positions must still round-trip + var events = CreateEvents(25).ToArray(); var streamName = GetStreamName(); await fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream, cancellationToken); var result = new List(); - await foreach (var evt in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 3, cancellationToken: cancellationToken)) { + await foreach (var evt in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 4, cancellationToken: cancellationToken)) { result.Add(evt); } @@ -68,6 +68,27 @@ public async Task ShouldReadStreamToEndAcrossPages(CancellationToken cancellatio await Assert.That(actual).IsEquivalentTo(events); } + [Test] + public async Task ShouldRejectLegacyUnrepresentableEntryId(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); + + // Entries written by older versions can carry auto-generated IDs with sequence numbers + // the position encoding can't represent; reading them must fail loudly, not garble positions + await fixture.GetDatabase().StreamAddAsync( + streamName.ToString(), + [ + new("message_id", Guid.NewGuid().ToString()), + new("message_type", serialized.EventType), + new("json_data", serialized.Payload), + new("created", DateTime.UtcNow.ToString(CultureInfo.InvariantCulture)) + ], + "12345-10" + ); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); + } + [Test] public async Task ShouldReturnEmptyReadingPastEnd(CancellationToken cancellationToken) { var events = CreateEvents(10).ToArray(); From 2d30d435130949dff59c7c4806dab644fb43455b Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 15:44:03 +0200 Subject: [PATCH 05/12] fix(redis): fail loudly on legacy entry IDs hidden behind a page boundary A legacy auto-generated entry ID with sequence 10+ sorts below the decoded form of the position that follows sequence 9, so a paged read could skip it before the read-side guard ever materialized it. Reads now probe the gap between the requested position and its decoded ID and throw NotSupportedException when unreachable legacy entries exist there. Also reverts the test fixture connection hardening, moved to a separate PR to keep this one focused. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 16 +++++++++ .../Fixtures/IntegrationFixture.cs | 4 +-- .../test/Eventuous.Tests.Redis/Store/Read.cs | 33 ++++++++++++++++--- 3 files changed, 46 insertions(+), 7 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 5cfb435b6..cefd6ce56 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -40,6 +40,22 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR StreamEvent[] events; try { + // Entries written by older versions with auto-generated IDs can sort below the decoded + // start position while falling inside the requested range: position m*10+s decodes to + // ID m-s, but a legacy entry (m-1)-(s+10) encodes to the same position or higher. + // Fail loudly when such entries exist instead of silently skipping them. + if (start.Value >= 10) { + var previousMs = start.Value / 10 - 1; + var hidden = await _getDatabase().StreamRangeAsync(stream.ToString(), $"{previousMs}-{start.Value % 10 + 10}", $"{previousMs}", count: 1).NoContext(); + + if (hidden is { Length: > 0 }) { + throw new NotSupportedException( + $"Redis stream entry ID {hidden[0].Id} can't be reached from position {start.Value}: the position encoding only supports ID sequence numbers 0-9. " + + "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store." + ); + } + } + // Range read is inclusive of the start position, matching the IEventReader contract // and the paged read extensions, which advance pages from the last revision + 1 var result = await _getDatabase().StreamRangeAsync(stream.ToString(), start.Value.ToRedisValue(), count: count).NoContext(); diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs index 0c37883ba..e237509d3 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs @@ -37,9 +37,7 @@ public async Task InitializeAsync() { IDatabase GetDb() { // FLUSHDB in test teardown is an admin command; StackExchange.Redis 3.x enforces the // admin gate for raw commands issued through Execute as well. - // abortConnect=false keeps the multiplexer retrying when the first connection attempt - // races the freshly started container. - var muxer = ConnectionMultiplexer.Connect($"{connString},allowAdmin=true,abortConnect=false"); + var muxer = ConnectionMultiplexer.Connect($"{connString},allowAdmin=true"); return muxer.GetDatabase(); } diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index d5f0d93e1..76dc92ea9 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -71,10 +71,37 @@ public async Task ShouldReadStreamToEndAcrossPages(CancellationToken cancellatio [Test] public async Task ShouldRejectLegacyUnrepresentableEntryId(CancellationToken cancellationToken) { var streamName = GetStreamName(); - var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); // Entries written by older versions can carry auto-generated IDs with sequence numbers // the position encoding can't represent; reading them must fail loudly, not garble positions + await AddLegacyEntry(streamName, "12345-10"); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); + } + + [Test] + public async Task ShouldRejectLegacyEntryIdHiddenBehindPageBoundary(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + // Legacy auto-generated IDs from a same-millisecond burst + for (var sequence = 0; sequence <= 10; sequence++) { + await AddLegacyEntry(streamName, $"12345-{sequence}"); + } + + // The first page ends at 12345-9 and the advanced position decodes past 12345-10, + // which must fail loudly instead of being silently skipped + await Assert.ThrowsAsync(ReadFunc); + + return; + + async Task ReadFunc() { + await foreach (var _ in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) { } + } + } + + async Task AddLegacyEntry(StreamName streamName, string id) { + var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); + await fixture.GetDatabase().StreamAddAsync( streamName.ToString(), [ @@ -83,10 +110,8 @@ await fixture.GetDatabase().StreamAddAsync( new("json_data", serialized.Payload), new("created", DateTime.UtcNow.ToString(CultureInfo.InvariantCulture)) ], - "12345-10" + id ); - - await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); } [Test] From 747be1cc22ffb00b4bcc1e02a19f61218c07b841 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 15:53:51 +0200 Subject: [PATCH 06/12] docs(redis): state the legacy position detection boundary Document that the read-side gap probe is complete for every position this store version can produce, and why positions minted by pre-fix versions from unrepresentable entries are inherently ambiguous (the encoding maps both legacy 12345-20 and valid 12347-0 to 123470). Add a test pinning that any read from the start of a legacy burst stream rejects the first unrepresentable entry. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 8 ++++++++ .../test/Eventuous.Tests.Redis/Store/Read.cs | 20 +++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index cefd6ce56..010a46176 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -44,6 +44,14 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR // start position while falling inside the requested range: position m*10+s decodes to // ID m-s, but a legacy entry (m-1)-(s+10) encodes to the same position or higher. // Fail loudly when such entries exist instead of silently skipping them. + // This check is complete for every position this store version can produce: revisions + // are only emitted for entries with sequence numbers 0-9, so the ID gap between a + // revision and the next position is exactly the range probed here, and older entries + // are materialized (and rejected) by the pages that precede the position. Positions + // minted by pre-fix versions from unrepresentable entries are inherently ambiguous — + // the encoding maps e.g. both legacy 12345-20 and valid 12347-0 to 123470 — and can't + // be detected without breaking reads of valid data; reading such streams from the + // start rejects the first unrepresentable entry. if (start.Value >= 10) { var previousMs = start.Value / 10 - 1; var hidden = await _getDatabase().StreamRangeAsync(stream.ToString(), $"{previousMs}-{start.Value % 10 + 10}", $"{previousMs}", count: 1).NoContext(); diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 76dc92ea9..373627ff6 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -99,6 +99,26 @@ async Task ReadFunc() { } } + [Test] + public async Task ShouldRejectLegacyBurstStreamReadFromStart(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + // A legacy burst with sequence numbers beyond a single decimal carry: positions minted for + // such entries by older versions are ambiguous, but any read from the start of the stream + // must reject the first unrepresentable entry it materializes + for (var sequence = 0; sequence <= 20; sequence += 5) { + await AddLegacyEntry(streamName, $"12345-{sequence}"); + } + + await Assert.ThrowsAsync(ReadFunc); + + return; + + async Task ReadFunc() { + await foreach (var _ in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, cancellationToken: cancellationToken)) { } + } + } + async Task AddLegacyEntry(StreamName streamName, string id) { var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); From 5934e2b3d9a5c31a18eb5e41110bcb3fdf93588d Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:01:53 +0200 Subject: [PATCH 07/12] fix(redis): reject resumed reads on streams with unrepresentable entry IDs A position minted by a pre-fix reader from a legacy entry with a multi-carry sequence number decodes past other legacy entries, which a resumed read then silently skipped. Since every entry that can hide from a position has a sequence number above 9, a stream is safe for resumed reads exactly when it holds no such entry. Reads from a non-zero position now verify that server-side (check_stream_clean function, clean verdict cached in a hash; entries written by the current store always carry sequence 0) and conservatively reject dirty streams with NotSupportedException naming the offending entry, replacing the single-carry gap probe. The caveat and the read-from-start migration guidance are documented on the public ReadEvents API. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 35 ++++++++++--------- .../Eventuous.Redis/Scripts/AppendEvents.lua | 31 +++++++++++++++- .../test/Eventuous.Tests.Redis/Store/Read.cs | 13 +++++++ 3 files changed, 61 insertions(+), 18 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 010a46176..7378eaa66 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -36,30 +36,31 @@ public RedisStore( const string ContentType = "application/json"; + /// + /// Reads events from a stream. Positions are inclusive of the start position. + /// Streams containing entries written by pre-0.16 versions with auto-generated IDs whose + /// sequence number exceeds 9 are not readable from a non-zero position: positions for such + /// entries don't round-trip through the position encoding, so resumed reads are rejected with + /// instead of risking silently skipped events. Read such + /// streams from the start, which fails loudly on the first unrepresentable entry, and migrate them. + /// public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { StreamEvent[] events; try { - // Entries written by older versions with auto-generated IDs can sort below the decoded - // start position while falling inside the requested range: position m*10+s decodes to - // ID m-s, but a legacy entry (m-1)-(s+10) encodes to the same position or higher. - // Fail loudly when such entries exist instead of silently skipping them. - // This check is complete for every position this store version can produce: revisions - // are only emitted for entries with sequence numbers 0-9, so the ID gap between a - // revision and the next position is exactly the range probed here, and older entries - // are materialized (and rejected) by the pages that precede the position. Positions - // minted by pre-fix versions from unrepresentable entries are inherently ambiguous — - // the encoding maps e.g. both legacy 12345-20 and valid 12347-0 to 123470 — and can't - // be detected without breaking reads of valid data; reading such streams from the - // start rejects the first unrepresentable entry. + // A resumed position is only unambiguous when every entry ID in the stream round-trips + // through the position encoding (sequence numbers 0-9). Entries the encoding can't + // represent can hide below the decoded start position while falling inside the + // requested range, so reads from a non-zero position are conservatively rejected for + // streams holding any such entry. The verdict is computed server-side and cached; + // entries written by the current store version always carry sequence 0. if (start.Value >= 10) { - var previousMs = start.Value / 10 - 1; - var hidden = await _getDatabase().StreamRangeAsync(stream.ToString(), $"{previousMs}-{start.Value % 10 + 10}", $"{previousMs}", count: 1).NoContext(); + var unclean = (string?)await _getDatabase().ExecuteAsync("FCALL", "check_stream_clean", 1, stream.ToString()).NoContext(); - if (hidden is { Length: > 0 }) { + if (!string.IsNullOrEmpty(unclean)) { throw new NotSupportedException( - $"Redis stream entry ID {hidden[0].Id} can't be reached from position {start.Value}: the position encoding only supports ID sequence numbers 0-9. " + - "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store." + $"Stream {stream} can't be read from position {start.Value}: it contains entry ID {unclean}, which the position encoding can't represent (only ID sequence numbers 0-9 are supported). " + + "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store. Read the stream from the start and migrate it." ); } } diff --git a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua index 80a2ab475..ac9a3626f 100644 --- a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua +++ b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua @@ -12,6 +12,34 @@ local function last_id_ms(key) return tonumber(string.sub(id, 1, string.find(id, '-', 1, true) - 1)) end +-- Returns the ID of the first entry whose sequence number can't be represented by the +-- client-side position encoding (sequence > 9), or an empty string when the stream is clean. +-- A clean verdict is cached: entries written by append_events always carry sequence 0, so a +-- stream verified clean stays clean. +local function check_stream_clean(keys, args) + local key = keys[1] + if redis.call('HGET', '_clean_streams', key) == '1' then + return '' + end + local cursor = '-' + while true do + local entries = redis.call('XRANGE', key, cursor, '+', 'COUNT', 1000) + if #entries == 0 then + break + end + for i=1,#entries do + local id = entries[i][1] + local seq = tonumber(string.sub(id, string.find(id, '-', 1, true) + 1)) + if seq > 9 then + return id + end + end + cursor = '(' .. entries[#entries][1] + end + redis.call('HSET', '_clean_streams', key, '1') + return '' +end + local function append_events(keys, args) local stream_name = keys[1] local expected_version = tonumber(keys[2]) @@ -67,4 +95,5 @@ local function append_events(keys, args) return {stream_version + items_inserted, global_position } end -redis.register_function('append_events', append_events) \ No newline at end of file +redis.register_function('append_events', append_events) +redis.register_function('check_stream_clean', check_stream_clean) \ No newline at end of file diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 373627ff6..2b49d6872 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -119,6 +119,19 @@ async Task ReadFunc() { } } + [Test] + public async Task ShouldRejectResumedCursorOnLegacyBurstStream(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + for (var sequence = 0; sequence <= 20; sequence++) { + await AddLegacyEntry(streamName, $"12345-{sequence}"); + } + + // A cursor minted by a pre-fix reader after consuming 12345-19 (revision 123469 + 1): + // resuming from it must be rejected, not silently skip the remaining entries + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); + } + async Task AddLegacyEntry(StreamName streamName, string id) { var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); From c15c9c4de2f623ccce83b8ff4de98cee67920f6b Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:19:10 +0200 Subject: [PATCH 08/12] fix(redis): make position validation incremental and its cache sound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the seventh review round: - Replace the server-side clean-stream function and its permanent name-keyed Redis marker with client-side validation in the store: the stream is scanned in bounded XRANGE pages that observe the caller's cancellation token, so the Redis server is never blocked for the length of the stream. - The verdict is cached per store instance, anchored on the first entry ID: Redis only accepts appends with increasing entry IDs, so a validated prefix can't gain entries, later reads only scan the delta above the last validated ID, a recreated stream (different first entry) triggers a full rescan, and a missing stream records no verdict — deleting, recreating, restoring, or importing legacy entries can no longer inherit a stale clean verdict. - Legacy-entry seeding in tests shares one connection handle instead of opening one per appended entry. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 77 ++++++++++++++++--- .../Eventuous.Redis/Scripts/AppendEvents.lua | 31 +------- .../test/Eventuous.Tests.Redis/Store/Read.cs | 56 ++++++++++++-- 3 files changed, 116 insertions(+), 48 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 7378eaa66..329ff99bf 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -1,6 +1,7 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. +using System.Collections.Concurrent; using System.Globalization; using System.Runtime.CompilerServices; using System.Runtime.Serialization; @@ -46,32 +47,25 @@ public RedisStore( /// public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { StreamEvent[] events; + var database = _getDatabase(); try { // A resumed position is only unambiguous when every entry ID in the stream round-trips // through the position encoding (sequence numbers 0-9). Entries the encoding can't // represent can hide below the decoded start position while falling inside the // requested range, so reads from a non-zero position are conservatively rejected for - // streams holding any such entry. The verdict is computed server-side and cached; - // entries written by the current store version always carry sequence 0. + // streams holding any such entry. if (start.Value >= 10) { - var unclean = (string?)await _getDatabase().ExecuteAsync("FCALL", "check_stream_clean", 1, stream.ToString()).NoContext(); - - if (!string.IsNullOrEmpty(unclean)) { - throw new NotSupportedException( - $"Stream {stream} can't be read from position {start.Value}: it contains entry ID {unclean}, which the position encoding can't represent (only ID sequence numbers 0-9 are supported). " + - "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store. Read the stream from the start and migrate it." - ); - } + await EnsureStreamPositionsRoundTrip(database, stream, cancellationToken).NoContext(); } // Range read is inclusive of the start position, matching the IEventReader contract // and the paged read extensions, which advance pages from the last revision + 1 - var result = await _getDatabase().StreamRangeAsync(stream.ToString(), start.Value.ToRedisValue(), count: count).NoContext(); + var result = await database.StreamRangeAsync(stream.ToString(), start.Value.ToRedisValue(), count: count).NoContext(); if (result == null! || result.Length == 0) { // An empty result can also mean the read window is past the stream end - if (!await _getDatabase().KeyExistsAsync(stream.ToString()).NoContext()) { + if (!await database.KeyExistsAsync(stream.ToString()).NoContext()) { throw new StreamNotFound(stream); } @@ -90,6 +84,65 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR public IAsyncEnumerable ReadEventsBackwards(StreamName stream, StreamReadPosition start, int count, CancellationToken cancellationToken) => throw new NotImplementedException(); + const int ValidationPageSize = 1000; + + readonly ConcurrentDictionary _validatedStreams = new(); + + // Validates that every entry ID in the stream round-trips through the position encoding. + // The verdict is cached per stream, anchored on the first entry ID: Redis only accepts + // appends with increasing entry IDs, so a validated range can't gain new entries, and a + // changed first entry means the stream was recreated. Only entries appended after the last + // validated one are scanned on subsequent reads, one bounded page at a time. + async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream, CancellationToken cancellationToken) { + var head = await database.StreamRangeAsync(stream, "-", "+", count: 1).NoContext(); + + // A missing stream holds nothing to validate, and no verdict is recorded for the name + if (head.Length == 0) return; + + var first = head[0].Id; + + RedisValue from; + RedisValue last; + + if (_validatedStreams.TryGetValue(stream, out var validated) && validated.First == first) { + from = $"({validated.Last}"; + last = validated.Last; + } else { + from = "-"; + last = first; + } + + while (true) { + cancellationToken.ThrowIfCancellationRequested(); + + var batch = await database.StreamRangeAsync(stream, from, "+", count: ValidationPageSize).NoContext(); + + if (batch.Length == 0) break; + + foreach (var entry in batch) { + if (EntrySequence(entry.Id) > 9) { + throw new NotSupportedException( + $"Stream {stream} can't be read from a non-zero position: it contains entry ID {entry.Id}, which the position encoding can't represent (only ID sequence numbers 0-9 are supported). " + + "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store. Read the stream from the start and migrate it." + ); + } + } + + last = batch[^1].Id; + from = $"({last}"; + + if (batch.Length < ValidationPageSize) break; + } + + _validatedStreams[stream] = (first, last); + } + + static long EntrySequence(RedisValue id) { + var value = Ensure.NotNull(id); + + return long.Parse(value.AsSpan(value.IndexOf('-') + 1)); + } + public async Task AppendEvents( StreamName stream, ExpectedStreamVersion expectedVersion, diff --git a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua index ac9a3626f..80a2ab475 100644 --- a/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua +++ b/src/Redis/src/Eventuous.Redis/Scripts/AppendEvents.lua @@ -12,34 +12,6 @@ local function last_id_ms(key) return tonumber(string.sub(id, 1, string.find(id, '-', 1, true) - 1)) end --- Returns the ID of the first entry whose sequence number can't be represented by the --- client-side position encoding (sequence > 9), or an empty string when the stream is clean. --- A clean verdict is cached: entries written by append_events always carry sequence 0, so a --- stream verified clean stays clean. -local function check_stream_clean(keys, args) - local key = keys[1] - if redis.call('HGET', '_clean_streams', key) == '1' then - return '' - end - local cursor = '-' - while true do - local entries = redis.call('XRANGE', key, cursor, '+', 'COUNT', 1000) - if #entries == 0 then - break - end - for i=1,#entries do - local id = entries[i][1] - local seq = tonumber(string.sub(id, string.find(id, '-', 1, true) + 1)) - if seq > 9 then - return id - end - end - cursor = '(' .. entries[#entries][1] - end - redis.call('HSET', '_clean_streams', key, '1') - return '' -end - local function append_events(keys, args) local stream_name = keys[1] local expected_version = tonumber(keys[2]) @@ -95,5 +67,4 @@ local function append_events(keys, args) return {stream_version + items_inserted, global_position } end -redis.register_function('append_events', append_events) -redis.register_function('check_stream_clean', check_stream_clean) \ No newline at end of file +redis.register_function('append_events', append_events) \ No newline at end of file diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 2b49d6872..6fa425fa0 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -1,6 +1,7 @@ using System.Globalization; using Eventuous.Tests.Redis.Fixtures; using Shouldly; +using StackExchange.Redis; using static Eventuous.Tests.Redis.Store.Helpers; namespace Eventuous.Tests.Redis.Store; @@ -74,7 +75,7 @@ public async Task ShouldRejectLegacyUnrepresentableEntryId(CancellationToken can // Entries written by older versions can carry auto-generated IDs with sequence numbers // the position encoding can't represent; reading them must fail loudly, not garble positions - await AddLegacyEntry(streamName, "12345-10"); + await AddLegacyEntry(fixture.GetDatabase(), streamName, "12345-10"); await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); } @@ -84,8 +85,10 @@ public async Task ShouldRejectLegacyEntryIdHiddenBehindPageBoundary(Cancellation var streamName = GetStreamName(); // Legacy auto-generated IDs from a same-millisecond burst + var database = fixture.GetDatabase(); + for (var sequence = 0; sequence <= 10; sequence++) { - await AddLegacyEntry(streamName, $"12345-{sequence}"); + await AddLegacyEntry(database, streamName, $"12345-{sequence}"); } // The first page ends at 12345-9 and the advanced position decodes past 12345-10, @@ -106,8 +109,10 @@ public async Task ShouldRejectLegacyBurstStreamReadFromStart(CancellationToken c // A legacy burst with sequence numbers beyond a single decimal carry: positions minted for // such entries by older versions are ambiguous, but any read from the start of the stream // must reject the first unrepresentable entry it materializes + var database = fixture.GetDatabase(); + for (var sequence = 0; sequence <= 20; sequence += 5) { - await AddLegacyEntry(streamName, $"12345-{sequence}"); + await AddLegacyEntry(database, streamName, $"12345-{sequence}"); } await Assert.ThrowsAsync(ReadFunc); @@ -123,8 +128,10 @@ async Task ReadFunc() { public async Task ShouldRejectResumedCursorOnLegacyBurstStream(CancellationToken cancellationToken) { var streamName = GetStreamName(); + var database = fixture.GetDatabase(); + for (var sequence = 0; sequence <= 20; sequence++) { - await AddLegacyEntry(streamName, $"12345-{sequence}"); + await AddLegacyEntry(database, streamName, $"12345-{sequence}"); } // A cursor minted by a pre-fix reader after consuming 12345-19 (revision 123469 + 1): @@ -132,10 +139,47 @@ public async Task ShouldRejectResumedCursorOnLegacyBurstStream(CancellationToken await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); } - async Task AddLegacyEntry(StreamName streamName, string id) { + [Test] + public async Task ShouldRejectResumedReadAfterStreamRecreatedWithLegacyEntries(CancellationToken cancellationToken) { + var events = CreateEvents(3).ToArray(); + var streamName = GetStreamName(); + await fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream, cancellationToken); + + // A resumed read on the clean stream passes validation + var appended = await fixture.EventReader.ReadEvents(streamName, new(10), 10, true, cancellationToken); + await Assert.That(appended.Length).IsGreaterThan(0); + + // Recreate the stream under the same name with legacy entries: the earlier verdict must not stick + var database = fixture.GetDatabase(); + await database.KeyDeleteAsync(streamName.ToString()); + + for (var sequence = 0; sequence <= 20; sequence++) { + await AddLegacyEntry(database, streamName, $"12345-{sequence}"); + } + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); + } + + [Test] + public async Task ShouldRejectResumedReadAfterMissingStreamGetsLegacyEntries(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + // A resumed read of a missing stream must not establish a verdict for the name + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(100), 10, true, cancellationToken)); + + var database = fixture.GetDatabase(); + + for (var sequence = 0; sequence <= 20; sequence++) { + await AddLegacyEntry(database, streamName, $"12345-{sequence}"); + } + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); + } + + static async Task AddLegacyEntry(IDatabase database, StreamName streamName, string id) { var serialized = EventSerializer.Default.SerializeEvent(CreateEvent()); - await fixture.GetDatabase().StreamAddAsync( + await database.StreamAddAsync( streamName.ToString(), [ new("message_id", Guid.NewGuid().ToString()), From abbbce2152f8e6b85c31ab97d9f043eba0c509b5 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:27:52 +0200 Subject: [PATCH 09/12] fix(redis): drop the position validation cache, validate per read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the eighth review round: no cache anchored on observable stream state is sound, because Redis exposes no immutable per-key generation identity — a stream restored with the same first entry defeated the first-entry anchor, and the cache grew unboundedly per stream name. Resumed reads now validate the prefix below the decoded position on every call, in bounded cancellable pages, scanning only entries the read itself won't materialize. Stateless validation can't go stale, holds no memory, and closes the head-read/scan TOCTOU; the per-read cost is documented on the public API. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 51 +++++++------------ .../test/Eventuous.Tests.Redis/Store/Read.cs | 22 ++++++++ 2 files changed, 40 insertions(+), 33 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 329ff99bf..aea73ff2e 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -1,7 +1,6 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. -using System.Collections.Concurrent; using System.Globalization; using System.Runtime.CompilerServices; using System.Runtime.Serialization; @@ -44,6 +43,9 @@ public RedisStore( /// entries don't round-trip through the position encoding, so resumed reads are rejected with /// instead of risking silently skipped events. Read such /// streams from the start, which fails loudly on the first unrepresentable entry, and migrate them. + /// To support that rejection, every read from a non-zero position validates the stream prefix + /// below the position in bounded batches, so resumed reads cost extra roundtrips proportional + /// to the prefix length. /// public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { StreamEvent[] events; @@ -56,7 +58,7 @@ public async IAsyncEnumerable ReadEvents(StreamName stream, StreamR // requested range, so reads from a non-zero position are conservatively rejected for // streams holding any such entry. if (start.Value >= 10) { - await EnsureStreamPositionsRoundTrip(database, stream, cancellationToken).NoContext(); + await EnsureStreamPositionsRoundTrip(database, stream, start, cancellationToken).NoContext(); } // Range read is inclusive of the start position, matching the IEventReader contract @@ -86,36 +88,22 @@ public IAsyncEnumerable ReadEventsBackwards(StreamName stream, Stre const int ValidationPageSize = 1000; - readonly ConcurrentDictionary _validatedStreams = new(); - - // Validates that every entry ID in the stream round-trips through the position encoding. - // The verdict is cached per stream, anchored on the first entry ID: Redis only accepts - // appends with increasing entry IDs, so a validated range can't gain new entries, and a - // changed first entry means the stream was recreated. Only entries appended after the last - // validated one are scanned on subsequent reads, one bounded page at a time. - async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream, CancellationToken cancellationToken) { - var head = await database.StreamRangeAsync(stream, "-", "+", count: 1).NoContext(); - - // A missing stream holds nothing to validate, and no verdict is recorded for the name - if (head.Length == 0) return; - - var first = head[0].Id; - - RedisValue from; - RedisValue last; - - if (_validatedStreams.TryGetValue(stream, out var validated) && validated.First == first) { - from = $"({validated.Last}"; - last = validated.Last; - } else { - from = "-"; - last = first; - } + // Validates that every entry ID below the decoded start position round-trips through the + // position encoding, scanning the current stream contents in bounded pages on every call. + // No verdict is cached: Redis has no immutable per-key generation identity, so a cached + // verdict can go stale when a key is deleted, recreated, or restored under the same name. + // Entries at or above the decoded position don't need validation here — the read + // materializes them, and converting an unrepresentable ID to a revision fails loudly. + // A key replaced concurrently with an in-flight read can still change underneath the scan, + // which no non-atomic paged read can detect; that also holds for the data reads themselves. + async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream, StreamReadPosition start, CancellationToken cancellationToken) { + RedisValue from = "-"; + var end = $"({start.Value.ToRedisValue()}"; while (true) { cancellationToken.ThrowIfCancellationRequested(); - var batch = await database.StreamRangeAsync(stream, from, "+", count: ValidationPageSize).NoContext(); + var batch = await database.StreamRangeAsync(stream, from, end, count: ValidationPageSize).NoContext(); if (batch.Length == 0) break; @@ -128,13 +116,10 @@ async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream } } - last = batch[^1].Id; - from = $"({last}"; - if (batch.Length < ValidationPageSize) break; - } - _validatedStreams[stream] = (first, last); + from = $"({batch[^1].Id}"; + } } static long EntrySequence(RedisValue id) { diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 6fa425fa0..53ce148f5 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -160,6 +160,28 @@ public async Task ShouldRejectResumedReadAfterStreamRecreatedWithLegacyEntries(C await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); } + [Test] + public async Task ShouldRejectResumedReadAfterStreamRestoredWithSameFirstEntry(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + var database = fixture.GetDatabase(); + + // A clean stream with explicit IDs, validated by a resumed read + await AddLegacyEntry(database, streamName, "12345-0"); + await AddLegacyEntry(database, streamName, "12346-0"); + await AddLegacyEntry(database, streamName, "12347-0"); + + var appended = await fixture.EventReader.ReadEvents(streamName, new(123460), 10, true, cancellationToken); + await Assert.That(appended.Length).IsGreaterThan(0); + + // Restore the stream with the same first entry but an unrepresentable entry + // below the previously validated range: the earlier verdict must not stick + await database.KeyDeleteAsync(streamName.ToString()); + await AddLegacyEntry(database, streamName, "12345-0"); + await AddLegacyEntry(database, streamName, "12346-10"); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); + } + [Test] public async Task ShouldRejectResumedReadAfterMissingStreamGetsLegacyEntries(CancellationToken cancellationToken) { var streamName = GetStreamName(); From 8e67ee4c8742961b4dbfe7509b3f19219a725954 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:34:25 +0200 Subject: [PATCH 10/12] fix(redis): honor the exception contract for unsigned ID sequences Redis stream ID components are unsigned 64-bit values; parsing the sequence with long.Parse raised OverflowException for sequences beyond long range instead of the documented NotSupportedException. Both the revision conversion and the validation scan now parse as ulong and range-check, and the millisecond part is range-checked before the signed conversion. Also document the operational requirement that pre-fix writers are quiesced before resumed reads are used: an old writer racing the gap between prefix validation and the data read can append an unrepresentable entry below the requested position, which only the next resumed read can reject. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 12 ++++++++++-- .../src/Eventuous.Redis/Tools/Conversions.cs | 16 ++++++++++++---- .../test/Eventuous.Tests.Redis/Store/Read.cs | 12 ++++++++++++ 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index aea73ff2e..5736017bb 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -46,6 +46,10 @@ public RedisStore( /// To support that rejection, every read from a non-zero position validates the stream prefix /// below the position in bounded batches, so resumed reads cost extra roundtrips proportional /// to the prefix length. + /// Pre-0.16 writers must be quiesced before resumed reads are used: an old writer racing the + /// gap between validation and the data read can append an unrepresentable entry below the + /// requested position, which that read won't see. Such a stream is rejected by the next + /// resumed read, but the racing read itself can't detect it. /// public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { StreamEvent[] events; @@ -96,6 +100,9 @@ public IAsyncEnumerable ReadEventsBackwards(StreamName stream, Stre // materializes them, and converting an unrepresentable ID to a revision fails loudly. // A key replaced concurrently with an in-flight read can still change underneath the scan, // which no non-atomic paged read can detect; that also holds for the data reads themselves. + // Likewise, a pre-fix writer appending an unrepresentable entry between this scan and the + // data read escapes the racing read (the next resumed read rejects the stream) — old writers + // must be quiesced before resumed reads are used, as documented on ReadEvents. async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream, StreamReadPosition start, CancellationToken cancellationToken) { RedisValue from = "-"; var end = $"({start.Value.ToRedisValue()}"; @@ -122,10 +129,11 @@ async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream } } - static long EntrySequence(RedisValue id) { + // Redis stream ID sequence components are unsigned 64-bit values + static ulong EntrySequence(RedisValue id) { var value = Ensure.NotNull(id); - return long.Parse(value.AsSpan(value.IndexOf('-') + 1)); + return ulong.Parse(value.AsSpan(value.IndexOf('-') + 1)); } public async Task AppendEvents( diff --git a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs index d44a6d98b..c77639e62 100644 --- a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs +++ b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs @@ -9,16 +9,24 @@ public static long ToLong(this RedisValue value) { return long.Parse(first) * 10 + long.Parse(second); } + // Redis stream ID components are unsigned 64-bit values, so both parts are parsed as ulong + // and range-checked before conversion to the signed position public static long ToRevision(this RedisValue value) { var (first, second) = new Split(Ensure.NotNull(value).AsSpan()); - var sequence = long.Parse(second); + var sequence = ulong.Parse(second); - return sequence <= 9 - ? long.Parse(first) * 10 + sequence - : throw new NotSupportedException( + if (sequence > 9) { + throw new NotSupportedException( $"Redis stream entry ID {value} can't be represented as a stream position: the position encoding only supports ID sequence numbers 0-9. " + "Entries with higher sequence numbers were written with auto-generated IDs by an older version of the store." ); + } + + var milliseconds = ulong.Parse(first); + + return milliseconds <= long.MaxValue / 10 + ? (long)milliseconds * 10 + (long)sequence + : throw new NotSupportedException($"Redis stream entry ID {value} can't be represented as a stream position: the millisecond part is too large."); } public static ulong ToULong(this ReadOnlySpan valueString) { diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 53ce148f5..8ad04bceb 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -102,6 +102,18 @@ async Task ReadFunc() { } } + [Test] + public async Task ShouldRejectEntryIdWithSequenceAboveLongRange(CancellationToken cancellationToken) { + var streamName = GetStreamName(); + + // Redis ID sequence components are unsigned 64-bit; values beyond long range must still + // surface as the documented NotSupportedException, both when materialized and when validated + await AddLegacyEntry(fixture.GetDatabase(), streamName, "12345-9223372036854775808"); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken)); + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); + } + [Test] public async Task ShouldRejectLegacyBurstStreamReadFromStart(CancellationToken cancellationToken) { var streamName = GetStreamName(); From 577b561edb141dc17437ae6bfa243867c143ae29 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:38:53 +0200 Subject: [PATCH 11/12] fix(redis): reject positions overflowing at the encoding boundary At milliseconds == long.MaxValue / 10 only sequences up to long.MaxValue % 10 fit the signed position; sequence 8 and 9 wrapped to a negative revision instead of throwing the documented NotSupportedException. Also widen the documented quiescence requirement to every writer that doesn't use this store version's explicit entry ID scheme, including external XADD with auto-generated IDs, not only pre-0.16 store versions. Co-Authored-By: Claude Fable 5 --- src/Redis/src/Eventuous.Redis/RedisStore.cs | 16 ++++++++++------ .../src/Eventuous.Redis/Tools/Conversions.cs | 7 +++++-- .../test/Eventuous.Tests.Redis/Store/Read.cs | 16 ++++++++++++++++ 3 files changed, 31 insertions(+), 8 deletions(-) diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs index 5736017bb..4a667af67 100644 --- a/src/Redis/src/Eventuous.Redis/RedisStore.cs +++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs @@ -46,10 +46,13 @@ public RedisStore( /// To support that rejection, every read from a non-zero position validates the stream prefix /// below the position in bounded batches, so resumed reads cost extra roundtrips proportional /// to the prefix length. - /// Pre-0.16 writers must be quiesced before resumed reads are used: an old writer racing the + /// Resumed reads require exclusive write ownership of the stream by this store version: any + /// writer that doesn't use its explicit entry ID scheme — a pre-0.16 store version, or any + /// external XADD with auto-generated IDs — must be quiesced first. Such a writer racing the /// gap between validation and the data read can append an unrepresentable entry below the - /// requested position, which that read won't see. Such a stream is rejected by the next - /// resumed read, but the racing read itself can't detect it. + /// requested position, which that read won't see. The stream is rejected by the next resumed + /// read, but the racing read itself can't detect it. Concurrent writers going through this + /// store version are safe. /// public async IAsyncEnumerable ReadEvents(StreamName stream, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) { StreamEvent[] events; @@ -100,9 +103,10 @@ public IAsyncEnumerable ReadEventsBackwards(StreamName stream, Stre // materializes them, and converting an unrepresentable ID to a revision fails loudly. // A key replaced concurrently with an in-flight read can still change underneath the scan, // which no non-atomic paged read can detect; that also holds for the data reads themselves. - // Likewise, a pre-fix writer appending an unrepresentable entry between this scan and the - // data read escapes the racing read (the next resumed read rejects the stream) — old writers - // must be quiesced before resumed reads are used, as documented on ReadEvents. + // Likewise, a writer not using the explicit entry ID scheme (a pre-fix store version or any + // external XADD with auto-generated IDs) appending an unrepresentable entry between this scan + // and the data read escapes the racing read (the next resumed read rejects the stream) — such + // writers must be quiesced before resumed reads are used, as documented on ReadEvents. async ValueTask EnsureStreamPositionsRoundTrip(IDatabase database, string stream, StreamReadPosition start, CancellationToken cancellationToken) { RedisValue from = "-"; var end = $"({start.Value.ToRedisValue()}"; diff --git a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs index c77639e62..f1c20ad9a 100644 --- a/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs +++ b/src/Redis/src/Eventuous.Redis/Tools/Conversions.cs @@ -24,9 +24,12 @@ public static long ToRevision(this RedisValue value) { var milliseconds = ulong.Parse(first); - return milliseconds <= long.MaxValue / 10 + const ulong maxMilliseconds = long.MaxValue / 10; + + // At the quotient boundary only sequences up to long.MaxValue % 10 still fit + return milliseconds < maxMilliseconds || (milliseconds == maxMilliseconds && sequence <= long.MaxValue % 10) ? (long)milliseconds * 10 + (long)sequence - : throw new NotSupportedException($"Redis stream entry ID {value} can't be represented as a stream position: the millisecond part is too large."); + : throw new NotSupportedException($"Redis stream entry ID {value} can't be represented as a stream position: the encoded value exceeds the position range."); } public static ulong ToULong(this ReadOnlySpan valueString) { diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index 8ad04bceb..d8f63d6d1 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -114,6 +114,22 @@ public async Task ShouldRejectEntryIdWithSequenceAboveLongRange(CancellationToke await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(streamName, new(123470), 10, true, cancellationToken)); } + [Test] + public async Task ShouldHandleRevisionBoundaryAtLongMax(CancellationToken cancellationToken) { + // long.MaxValue / 10 = 922337203685477580, long.MaxValue % 10 = 7: sequence 7 encodes to + // exactly long.MaxValue, sequence 8 no longer fits and must be rejected, not wrap negative + var fitting = GetStreamName(); + await AddLegacyEntry(fixture.GetDatabase(), fitting, "922337203685477580-7"); + + var result = await fixture.EventReader.ReadEvents(fitting, StreamReadPosition.Start, 10, true, cancellationToken); + await Assert.That(result[0].Revision).IsEqualTo(long.MaxValue); + + var overflowing = GetStreamName(); + await AddLegacyEntry(fixture.GetDatabase(), overflowing, "922337203685477580-8"); + + await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(overflowing, StreamReadPosition.Start, 10, true, cancellationToken)); + } + [Test] public async Task ShouldRejectLegacyBurstStreamReadFromStart(CancellationToken cancellationToken) { var streamName = GetStreamName(); From 9b7fe624695ba2fc03d89375efed3f7b0fd104ad Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 19 Aug 2026 16:43:09 +0200 Subject: [PATCH 12/12] fix(persistence): stop paged reads advancing past the maximum revision An event at revision long.MaxValue that fills an exact page made ReadStreamToEnd compute lastRevision + 1, wrapping negative and throwing from the StreamReadPosition constructor after yielding the event. The maximum revision is the end of the representable position space, so the paged read now completes there. Co-Authored-By: Claude Fable 5 --- .../EventStore/StoreFunctions.cs | 3 +++ .../test/Eventuous.Tests.Redis/Store/Read.cs | 17 +++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs index aaa226078..18a5f430f 100644 --- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs +++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs @@ -240,6 +240,9 @@ [EnumeratorCancellation] CancellationToken cancellationToken if (yielded < pageSize) yield break; + // The maximum revision is the end of the representable position space + if (lastRevision == long.MaxValue) yield break; + position = new(lastRevision + 1); } } diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs index d8f63d6d1..4269017a0 100644 --- a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs +++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs @@ -130,6 +130,23 @@ public async Task ShouldHandleRevisionBoundaryAtLongMax(CancellationToken cancel await Assert.ThrowsAsync(() => fixture.EventReader.ReadEvents(overflowing, StreamReadPosition.Start, 10, true, cancellationToken)); } + [Test] + public async Task ShouldReadStreamToEndAtMaxRevision(CancellationToken cancellationToken) { + // An event at the maximum representable revision filling an exact page must complete + // the paged read instead of advancing past the end of the position space + var streamName = GetStreamName(); + await AddLegacyEntry(fixture.GetDatabase(), streamName, "922337203685477580-7"); + + var result = new List(); + + await foreach (var evt in fixture.EventReader.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 1, cancellationToken: cancellationToken)) { + result.Add(evt); + } + + await Assert.That(result).HasCount().EqualTo(1); + await Assert.That(result[0].Revision).IsEqualTo(long.MaxValue); + } + [Test] public async Task ShouldRejectLegacyBurstStreamReadFromStart(CancellationToken cancellationToken) { var streamName = GetStreamName();