Azure blob storage projection - #550
Conversation
…StorageBlobsProjector - Create new test project following Eventuous.Tests.Azure.ServiceBus structure - Add Testcontainers.Azurite package to Directory.Packages.props - Add IntegrationFixture with Azurite and KurrentDB containers - Test all On method variants (sync/async, state/context) for new and existing blobs - Test concurrent modification scenario (412 Precondition Failed) - Test no handler scenario (returns Ignored) Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
…face intent - Add helper methods: SetupContainer, SetupExistingBlob, GetBlobState, AssertSuccess, AssertIgnored - Rename projector classes to surface handler patterns (SyncStateProjector, etc.) - Group tests by handler variant with clear section comments - Test names now follow [Variant]_[Scenario]_[ExpectedBehavior] pattern - Reduce LOC from ~450 to ~330 (-27%) - Remove fixture parameter from CreateContext (unused) Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
…obServiceClient and container name - Add StorageBlobsProjector(BlobServiceClient, string containerName) constructor - Update test helper methods to work with container names instead of BlobContainerClient - Add GetContainer() helper to get BlobContainerClient from fixture - Update all test projector classes with new constructor overload - Update all tests to use fixture.BlobServiceClient with container names Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
PR Summary by QodoAdd Azure Blob Storage projector and .NET Aspire Azure sample Description
Diagram
High-Level Assessment
Files changed (39)
|
Code Review by Qodo
1.
|
![qodo-free-for-open-source-projects[bot]](https://avatars.githubusercontent.com/in/1068000?s=60&v=4){kind=small}
Test Results 46 files + 24 46 suites +24 11m 57s ⏱️ -34s Results for commit 4edc5c9. ± Comparison against base commit 3cb68c2. This pull request removes 5 and adds 63 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
|
Thanks so much for this @quezlatch — really nice work, and I appreciate you building it on top of your Service Bus piece. The blob projector is clean, the ETag-based optimistic concurrency is exactly right, and I love that you backed it with proper Azurite testcontainer tests covering the concurrent-modification cases. The Aspire sample is genuinely cool too — distributed debugging "just working" is a great thing to show off. 🙂 A few things before I can merge, mostly housekeeping rather than anything wrong with the code:
Couple of tiny optional nits: the None of this is a big deal — the hard part is done and it's looking great. Thanks again for contributing! 🙏 |
|
I will take a stab at the docs as well. Assuming it should go in the |
quezlatch
changed the title
|
#556 is stacked on top of this |
alexeyzimarev
left a comment
alexeyzimarev
left a comment
There was a problem hiding this comment.
Thanks for the update. I found one correctness issue that blocks the idempotency guarantee, plus a packed README type error. I reproduced the idempotency issue with two Azurite regression tests; both modes returned Success for a non-consecutive duplicate. Please address the inline comments.
|
I've addressed those issues. Hopefully it looks better now. I've also updated the docs and merged to the stacked #556 PR |
Eventuous#568) * 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 Eventuous#567 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * fix(persistence): bound tiered reads and make Redis reads inclusive 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * fix(redis): make position validation incremental and its cache sound 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 <noreply@anthropic.com> * fix(redis): drop the position validation cache, validate per read 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
A read that faults or is cancelled before the first response parks the failure in two places: the message channel, which the enumerator observes, and the public ReadState task, which nothing awaits. The faulted ReadState task was then collected unobserved and surfaced on the finalizer thread as a TaskScheduler.UnobservedTaskException. Every read is affected, not just cancelled ones: the leak window is "fault before the first response", so connection failures, auth failures, deadline expiry and server unavailability all leak too. Client 1.4.1 observes the fault at the source, and also fixes two sibling sinks that no consumer can reach from outside: SharingProvider's call-invoker boxes and the batch appender's fire-and-forget send loop. Measured over 20 reads that fail before the first response, against a closed port: 59 unobserved exceptions on 1.4.0 (18 from ReadState, 21 from SharingProvider retries, 20 from disposed boxes), 0 on 1.4.1. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ventuous#572) * fix(test): stop Redis fixture aborting on first connection attempt The Redis test fixture connects right after the container reports ready, and the first connection attempt occasionally races the server, failing the whole fixture initialization with RedisConnectionException before any test runs. abortConnect=false makes the multiplexer keep retrying instead of aborting. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Connect the Redis test fixture asynchronously with a shared multiplexer Address the review note: replace the synchronous ConnectionMultiplexer.Connect inside InitializeAsync with an awaited ConnectAsync. Connect once and share the multiplexer across tests instead of opening a new connection per GetDatabase call, and dispose it with the fixture; abortConnect=false is preserved so the first attempt keeps retrying when it races the freshly started container. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
- Remove leftovers from the extracted Aspire sample: orphan package versions and an unrelated ServiceBusSubscription signature change - Consolidate JSON serialization config into BlobStorageProjectorOptions.JsonOptions, dropping the constructor serializerOptions parameter - Warn when ByGlobalPosition idempotency receives events with global position 0, and document that the mode requires real global positions - Add copyright headers, follow .editorconfig accessibility and naming conventions, inline the misnamed GetBlobContainerClient, drop the redundant On<TEvent> overload and the manual ValueTask fast-path - Remove dead event store scaffolding and KurrentDB references from the test project, dedupe concurrent-modification test lambdas - Fix README: stale IOptions claim, blob naming example, container existence requirement Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- Catch 404 only for the projection blob read (filtered to BlobNotFound) and 412/409 only for the conditional upload (filtered to ConditionNotMet/BlobAlreadyExists), so exceptions thrown by the user-supplied event handler are never misclassified as ETag races or missing-blob conditions and the handler is never re-invoked for them - Percent-encode Stream and MessageId blob metadata values: Azure requires ASCII metadata, while stream names and message ids can be arbitrary strings (e.g. Booking-Ålesund previously failed uploads with InvalidMetadata) - Add tests for both: handler-thrown RequestFailedException propagates without retries, and Unicode stream names project successfully with encoded metadata Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Deserialization and upload preparation (JSON serialization through the user-configurable options, upload options and metadata construction) now run outside the try blocks, so each catch classifies exclusively its own SDK call: DownloadContentAsync for the missing-blob path and UploadAsync for the concurrency-conflict path. Exceptions from user-supplied JSON converters can no longer be misread as blob conditions. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
alexeyzimarev
left a comment
alexeyzimarev
left a comment
There was a problem hiding this comment.
Cleanup applied and independent review passed (3 rounds, clean). Approving to supersede my earlier change request.
3 participants
Azure blob storage
Adds a new projection target for persisting state to Azure Blob Storage. The implementation provides a flexible way to project events into blob-stored state objects.
PR prepared with the help of Mistral Vibe.