fix: true streaming reads in KurrentDB store and paged read-to-end API

src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs

@@ -7,6 +7,12 @@ public interface IEventReader {
     /// <summary>
     /// Read a fixed number of events from an existing stream as an async enumerable.
     /// Throws <see cref="StreamNotFound"/> if the stream does not exist.
+    /// Implementations either stream events as they arrive from the store, or buffer events in an amount
+    /// proportional to <paramref name="count"/> before yielding, so memory usage can grow with
+    /// <paramref name="count"/>. To read a whole stream, use <see cref="StoreFunctions.ReadStreamToEnd"/>,
+    /// which reads in pages, instead of passing <see cref="int.MaxValue"/> as the count.
+    /// Implementations must yield exactly <paramref name="count"/> 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.
     /// </summary>
     /// <param name="stream">Stream name</param>
     /// <param name="start">Where to start reading events</param>
@@ -18,6 +24,9 @@ public interface IEventReader {
     /// <summary>
     /// Read a number of events from a given stream, backwards (from the stream end).
     /// Throws <see cref="StreamNotFound"/> if the stream does not exist.
+    /// Implementations either stream events as they arrive from the store, or buffer events in an amount
+    /// proportional to <paramref name="count"/> before yielding, so memory usage can grow with
+    /// <paramref name="count"/>.
     /// </summary>
     /// <param name="stream">Stream name</param>
     /// <param name="start">Where to start reading events</param>

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,33 @@ CancellationToken  cancellationToken
             }
         }
 
+        /// <summary>
+        /// Reads a stream from the given position to the end, as an async enumerable.
+        /// Events are read in pages of <paramref name="pageSize"/> and yielded as they arrive, so the whole stream
+        /// is never buffered in memory. Use this instead of calling <see cref="IEventReader.ReadEvents"/>
+        /// with <see cref="int.MaxValue"/> as the count.
+        /// </summary>
+        /// <param name="streamName">Name of the stream to read from</param>
+        /// <param name="start">Stream position to start reading from</param>
+        /// <param name="pageSize">Number of events to read per page. It bounds the memory a buffering
+        /// implementation of <see cref="IEventReader"/> 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).</param>
+        /// <param name="failIfNotFound">Set to false to complete without yielding anything when the stream isn't found,
+        /// instead of throwing <see cref="StreamNotFound"/>. Default is true.</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        /// <returns>An async enumerable of events retrieved from the stream</returns>
+        public IAsyncEnumerable<StreamEvent> ReadStreamToEnd(
+                StreamName        streamName,
+                StreamReadPosition start,
+                int               pageSize          = 500,
+                bool              failIfNotFound    = true,
+                CancellationToken cancellationToken = default
+            ) {
+            ArgumentOutOfRangeException.ThrowIfNegativeOrZero(pageSize);
+
+            return ReadToEnd(eventReader, streamName, start, pageSize, failIfNotFound, cancellationToken);
+        }
+
         /// <summary>
         /// Reads a stream from the event store to a collection of <seealso cref="StreamEvent"/>
         /// </summary>
@@ -163,26 +192,58 @@ public async Task<StreamEvent[]> ReadStream(
                 bool               failIfNotFound    = true,
                 CancellationToken  cancellationToken = default
             ) {
-            const int pageSize = 500;
-
             var streamEvents = new List<StreamEvent>();
 
-            var position = start;
-
-            try {
-                while (true) {
-                    var events = await eventReader.ReadEvents(streamName, position, pageSize, failIfNotFound, cancellationToken).NoContext();
-                    streamEvents.AddRange(events);
+            await foreach (var evt in eventReader.ReadStreamToEnd(streamName, start, failIfNotFound: failIfNotFound, cancellationToken: cancellationToken).NoContext(cancellationToken)) {
+                streamEvents.Add(evt);
+            }
 
-                    if (events.Length < pageSize) break;
+            return [.. streamEvents];
+        }
+    }
 
-                    position = new(position.Value + events.Length);
+    // 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<StreamEvent> 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;
                 }
-            } catch (StreamNotFound) when (!failIfNotFound) {
-                return [];
+
+                if (!moved) break;
+
+                var evt = enumerator.Current;
+                yielded++;
+                lastRevision = evt.Revision;
+
+                yield return evt;
             }
 
-            return [.. streamEvents];
+            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);
         }
     }
 }

src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs

@@ -13,16 +13,29 @@ namespace Eventuous;
 /// <param name="archiveReader">Event reader pointing to archive store</param>
 public class TieredEventReader(IEventReader hotReader, IEventReader archiveReader) : IEventReader {
     public async IAsyncEnumerable<StreamEvent> 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<StreamEvent> archivedEvents;
+        var                      archiveNotFound = false;
+
+        switch (hotEvents.Length) {
+            case > 0 when hotEvents[0].Revision > start.Value: {
+                // 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;
+            }
+            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 combined = archivedEvents.Concat(hotEvents).Distinct(Comparer).Take(count);
         var any      = false;
 
         foreach (var evt in combined) {
@@ -31,28 +44,32 @@ public async IAsyncEnumerable<StreamEvent> 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<StreamEvent> 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<StreamEvent> archivedEvents;
+        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;
 
-                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 +83,12 @@ public async IAsyncEnumerable<StreamEvent> 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<StreamEvent[]> LoadStreamEvents(
+    static async Task<(StreamEvent[] Events, bool NotFound)> LoadStreamEvents(
             IEventReader       reader,
             StreamName         streamName,
             StreamReadPosition startPosition,
@@ -78,11 +97,13 @@ static async Task<StreamEvent[]> 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);
         }
     }
 

src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs

@@ -146,6 +146,117 @@ 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<StreamNotFound>(() => _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<StreamNotFound>(() => _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<StreamEvent>();
+
+        await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) {
+            result.Add(evt);
+        }
+
+        IEnumerable<object> 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<StreamEvent>();
+
+        await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) {
+            result.Add(evt);
+        }
+
+        IEnumerable<object> 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<StreamEvent>();
+
+        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 ShouldRejectInvalidPageSizeReadingToEnd(CancellationToken cancellationToken) {
+        var streamName = Helpers.GetStreamName();
+
+        await Assert.ThrowsAsync<ArgumentOutOfRangeException>(() => Read(0));
+        await Assert.ThrowsAsync<ArgumentOutOfRangeException>(() => 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) {
+        var streamName = Helpers.GetStreamName();
+
+        await Assert.ThrowsAsync<StreamNotFound>(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<StreamEvent>();
+
+        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) {