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");