Skip to content

feat: streaming query pipeline (IAsyncEnumerable, ExecuteStream, resilience) — closes #299#350

Merged
iancooper merged 37 commits into
masterfrom
feat/299-streaming
Jul 13, 2026
Merged

feat: streaming query pipeline (IAsyncEnumerable, ExecuteStream, resilience) — closes #299#350
iancooper merged 37 commits into
masterfrom
feat/299-streaming

Conversation

@iancooper

Copy link
Copy Markdown
Member

Summary

Implements spec 012 — streaming query results — across 31 tasks and 9 phases (T001–T031).

  • New IStreamQuery<TResult> marker and IStreamQueryHandler<TQuery,TResult> base contract with StreamQueryHandler<,> abstract base; TResult is the item type, not the enumerable
  • IQueryProcessor.ExecuteStream<TResult> — async iterator entry point; span + handler/decorator lifetime are bound to enumeration so early break and exceptions never leak resources
  • PipelineBuilder.BuildStream — resolves handlers from a dedicated StreamQueryHandlerRegistry, chains IStreamQueryHandlerDecorator instances ordered by Step descending, validates that legacy sync/async attributes are not used on stream handlers (and vice versa)
  • Stream logging decorator (src/Paramore.Darker/Logging/) — logs start, yields items lazily, records item count + duration on completion, captures mid-stream faults
  • UseResiliencePipelineStreamHandler (Polly v8) — wraps stream establishment and the first MoveNextAsync inside the named pipeline; items are yielded only after the pipeline succeeds, so retry produces no duplicate emission; faults after the first item propagate un-retried; failed-attempt enumerators are disposed before rethrowing
  • DI wiringServiceCollectionStreamHandlerRegistry, AddHandlersFromAssemblies extended to scan IStreamQueryHandler<,>, new AddStreamHandlers(Action<IStreamQueryHandlerRegistry>) builder method, HandlerConfiguration carries the stream registry into QueryProcessor
  • QueryProcessorBuilder — accepts a HandlerConfiguration with a stream registry, no additional changes required (T006/T008 already covered this)
  • FakeQueryProcessor updated with a no-op ExecuteStream implementation
  • netstandard2.0 support via conditional Microsoft.Bcl.AsyncInterfaces package reference
  • AOT — no new trim/AOT-hostile reflection beyond the existing BuildAsync path; AOT publish verified green
  • Documentation — new "Streaming Queries" section in README.md covering usage, DI registration, resilience, and all documented semantics

Documented semantics (key behavioural differences from ExecuteAsync)

Behaviour Detail
Laziness Framework never buffers; items produced on demand
Config errors Deferred to first await foreach iteration (deliberate — avoids handler leak when caller never enumerates)
Re-enumeration Re-executes the handler with a fresh pipeline each time (cold, not cached)
Resilience scope Covers establishment + first item only; Timeout bounds start-up; Hedging unsupported
Legacy attributes [RetryableQuery]/[FallbackPolicy] throw ConfigurationException on a stream handler; use [UseResiliencePipelineStream]
Caller context A caller-supplied queryContext is scoped to a single enumeration

Test plan

  • 201 core tests pass (net8.0 + net9.0) covering: laziness, cancellation mid-stream, exceptions mid-stream, early-break lifetime release, deferred config error, decorator ordering, mismatch rejection, re-enumeration, logging lifecycle, span events per step, resilience happy path, retry-no-duplicate, post-first-item fault propagation, enumerator disposal on retry, fallback stream substitution
  • 39 DI extension tests pass covering: assembly scan registering stream handlers, explicit AddStreamHandlers, QueryProcessorBuilder with stream registry
  • Release build clean (0 errors, 305 pre-existing warnings)
  • AOT publish (net9.0, osx-arm64) succeeds with no new IL warnings beyond pre-existing BuildAsync ones

Closes #299

🤖 Generated with Claude Code

iancooper and others added 30 commits July 9, 2026 08:21
Add requirements, design review, and ADR 0019 for the streaming
query pipeline feature (spec 012-streaming_results).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add tasks.md with 31 TDD-first tasks across 9 phases, sequenced
structural-before-behavioural per ADR 0019, with a dependency graph
and risk-mitigation traceability table. Update spec README checklist.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…003)

Add IAsyncEnumerable support via Microsoft.Bcl.AsyncInterfaces (netstandard2.0 only),
stream query/handler/decorator contracts, and the StreamQueryHandlerAttribute base.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ry type (T004)

Adds IStreamQueryHandlerRegistry and StreamQueryHandlerRegistry with generic/type-based
registration, duplicate and result-type guards, and RegisterFromAssemblies scan.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…plementations (T005)

Assembly scan was already implemented in T004; these tests lock in the behavior:
stream handlers are discovered, async/sync handlers are ignored.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…er, and processor (T006)

Additive structural wiring: HandlerConfiguration gains optional StreamHandlerRegistry member,
PipelineBuilder gains a new ctor slot, QueryProcessor reads it from configuration.
No behaviour change.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…teStream (T007)

BuildStream resolves the stream handler by signature (not bare name) to avoid
AmbiguousMatchException; ExecuteStream is an async iterator binding span and
pipeline lifetime to enumeration. FakeQueryProcessor updated for new interface.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… provided context (T008)

ExecuteStream was implemented in T007; these tests lock in the full wiring behavior
including cancellation token flow and caller-supplied context preservation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ffering (T009)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… OperationCanceledException (T010)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…vocationException (T011)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…y once (T012)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…Exception (T013)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…c for unregistered stream query (T014)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…g (T015)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…async pipelines (T016)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…eline (T017)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… and duration (T018)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…level (T019)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… (T020)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ndler stub (T021)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…lishment succeeds yields all items (T022)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… duplicate emission (T023)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…024)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…umerator (T025)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
iancooper and others added 7 commits July 13, 2026 12:59
… at establishment (T026)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Add ServiceCollectionStreamHandlerRegistry that registers stream handler
types in DI alongside the sync/async registries; thread it through
AddHandlersFromAssemblies and BuildQueryProcessor into HandlerConfiguration
so ExecuteStream resolves handlers from the container after an assembly scan.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Add AddStreamHandlers(Action<IStreamQueryHandlerRegistry>) to
IDarkerHandlerBuilder and implement it on ServiceCollectionDarkerHandlerBuilder,
mirroring AddAsyncHandlers so callers can explicitly register stream handlers
without an assembly scan.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Verify that a processor built via the fluent QueryProcessorBuilder with a
StreamQueryHandlerRegistry in HandlerConfiguration executes stream queries
via ExecuteStream; behaviour was already covered by T006/T008 so no new
implementation was required.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
T027-T029 DI wiring complete; T030 verified green on net8.0/net9.0
Release build and AOT publish with no new warnings or errors.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Add a 'Streaming Queries' section to the root README covering:
ExecuteStream + await foreach usage, DI registration (scan and explicit),
QueryProcessorBuilder wiring, UseResiliencePipelineStream attribute, and
a semantics table (laziness, cancellation, exceptions, deferred config
error, re-enumeration, caller context, legacy attribute restriction).

Update specs/012-streaming_results/README.md with full implementation
summary — all 31 tasks across 9 phases complete.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@iancooper iancooper self-assigned this Jul 13, 2026
@iancooper iancooper added enhancement .NET Pull requests that update .net code 3 - Done V5 labels Jul 13, 2026
@iancooper iancooper merged commit 8120f8b into master Jul 13, 2026
4 checks passed
@iancooper iancooper deleted the feat/299-streaming branch July 13, 2026 20:23
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

3 - Done enhancement .NET Pull requests that update .net code V5

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add streaming query support with IAsyncEnumerable

1 participant