fix: true streaming reads in KurrentDB store and paged read-to-end API - #568
fix: true streaming reads in KurrentDB store and paged read-to-end API#568alexeyzimarev wants to merge 1 commit into
Conversation
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 <noreply@anthropic.com>
PR Summary by QodoTrue streaming reads for KurrentDB and paged ReadStreamToEnd API
AI Description
Diagram
High-Level Assessment
Files changed (8)
|
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| } 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); | ||
| } |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 12c0020b87
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| public async IAsyncEnumerable<StreamEvent> ReadStreamToEnd( | ||
| StreamName streamName, | ||
| StreamReadPosition start, | ||
| int pageSize = 500, |
There was a problem hiding this comment.
When pageSize is zero, readers such as SqlEventStoreBase return an empty page, but yielded < pageSize is false, so the outer loop repeats indefinitely and continuously queries the store until cancellation. Negative values can behave similarly for readers that treat non-positive counts as empty. Validate that this public argument is greater than zero before entering the paging loop.
Useful? React with 👍 / 👎.
| yield return evt; | ||
| } | ||
|
|
||
| if (yielded < pageSize) yield break; |
There was a problem hiding this comment.
When a KurrentDB page contains an event that EnumerateStream suppresses, such as an unresolved $> link event whose deserialization fails, the enumerable yields fewer than pageSize items even though the underlying raw page was full and later pages exist. Treating the number of yielded user events as proof that the source reached its end therefore makes ReadStreamToEnd silently omit the remaining events; page exhaustion must be tracked independently of filtered events.
Useful? React with 👍 / 👎.
Code Review by Qodo
1. Infinite loop on pageSize
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
| if (yielded < pageSize) yield break; | ||
|
|
||
| position = new(lastRevision + 1); | ||
| } |
There was a problem hiding this comment.
1. Infinite loop on pagesize 🐞 Bug ☼ Reliability
StoreFunctions.ReadStreamToEnd can loop forever when pageSize <= 0 because it only exits when yielded < pageSize, which is false for yielded == 0 and pageSize <= 0, so it keeps advancing position and issuing empty reads.
Agent Prompt
### Issue description
`ReadStreamToEnd` does not validate `pageSize`. When `pageSize <= 0`, the paging loop can become non-terminating (especially for implementations that yield no events for `count <= 0`), repeatedly performing empty reads.
### Issue Context
This method is a new public read-to-end API intended for safe, bounded-memory paging. Invalid `pageSize` values should fail fast (or be normalized) to prevent hangs.
### Fix Focus Areas
- src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs[167-204]
### Suggested fix
- Add an argument guard at the start of `ReadStreamToEnd`, e.g.:
- `if (pageSize <= 0) throw new ArgumentOutOfRangeException(nameof(pageSize), "pageSize must be > 0");`
- (Optional) Add a regression test ensuring `pageSize: 0` (and negative) throws `ArgumentOutOfRangeException`.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Test Results 46 files + 24 46 suites +24 12m 2s ⏱️ -29s Results for commit 12c0020. ± Comparison against base commit 3cb68c2. This pull request removes 5 and adds 39 tests. Note that renamed tests count towards both. |
1 participant
Closes #567
Summary
KurrentDBEventStore:ReadEventsandReadEventsBackwardsnow yield each event as it arrives from the server instead of materializing the whole requested range into two stream-sized arrays before the first yield. Exception mapping (StreamNotFound,ReadFromStreamException) and logging are preserved by wrapping each advance of the source enumerator;OperationCanceledExceptionnow propagates instead of being wrapped.IEventReader.ReadStreamToEndextension returnsIAsyncEnumerable<StreamEvent>, reading in pages (default 500, tunable) and advancing from the last yielded event's revision — bounded memory on every provider, socount: int.MaxValuestops being the idiom.ReadStreamnow delegates to it, which also fixes its page advancement for truncated streams.IEventReader.ReadEvents/ReadEventsBackwards: implementations either stream or buffer up tocount, and whole-stream reads should useReadStreamToEnd.New shared contract tests exposed two pre-existing provider bugs, fixed here:
StreamNotFoundwhen reading a missing stream (a plain SELECT can't tell a missing stream from a read past the end).SqlEventStoreBasenow checksStreamExistswhen a read returns no rows.StreamReadPosition.End:long.MaxValuewas marshalled into an INT parameter (arithmetic overflow). The parameter is now clamped to the 32-bit position range, which is lossless since both schemas store positions as INT and both procedures already trim the position to the stream head.Test plan
StreamingReadTests(KurrentDB) prove streaming with a counting serializer: exactly 1 event deserialized at first yield, previously the full rangeStoreReadTestscases (inherited by KurrentDB, Postgres, SqlServer, Sqlite): read-to-end across page boundaries, exact page multiples, from a position, missing-stream throw/empty behavior, and missing-stream contract for plain reads🤖 Generated with Claude Code