From 12c0020b878677a7b3b9be3eac93033aa227c6e9 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 12 Aug 2026 18:47:52 +0200 Subject: [PATCH] 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");