[ledjon-behluli]

This PR adds support for IAsyncEnumerable<T> returning methods in grain observers. So infos that might be worth pointing out:

• Support for IAE in observer methods is done through a new AsyncEnumerableGrainObserverExtension, owned by LocalObjectData.
• The rest of the pipeline is shared with what grains use, including IAsyncEnumerableGrainExtension, IAsyncEnumerableRequest, and code generation, so we don't duplicate the existing request/response pipeline.
• The main difference is that grains have their own execution context and scheduling guarantees, with observers its much harder to get right since observer calls, cleanup, deregistration, and shutdown can race independently through LocalObjectData.
• Due to the above reason the observer extension needs explicit synchronization through an extension-level lock and per-enumerator gates, ensuring MoveNextAsync and DisposeAsync don't overlap and stale requests can't operate on replaced enumerators.
• In addition I added joinable LocalObjectData disposal and tracking of deregistration-triggered disposals so InvokableObjectManager.StopAsync doesn't miss cleanup already in progress once.
• I aslo added ObserverNotReachableException when Dispatch can't find the local observer anymore, returning it for request-response messages while preserving the existing OneWay message behavior.
• Tests have parity (and them some) with the existing extension, and also the changes outside the main thing such as the new Dispatch behavior and joinable disposals in the manager are also covered.

Microsoft Reviewers: Open in CodeFlow

[Copilot]

🟡 Changes recommended

Unresolved concurrency races can silently drop producer results and block StopAsync indefinitely, while explicit disposal can return before producer teardown completes.

5 open issues

• Signal observer removal independently of blocked teardown · New
• Preserve completed producer task when returning Heartbeat · New
• Make explicit disposal await producer teardown · New
• Add required Suite and Provider traits to observer tests · New
• Join producer before deregistering the observer · New

What changed in this PR

Adds support for observer methods returning IAsyncEnumerable<T>, reusing Orleans’ existing enumeration protocol while managing observer-specific concurrency and cleanup.

Changes:

• Adds an observer enumeration extension with synchronized iteration, cancellation, and disposal.
• Tracks observer cleanup during deregistration and shutdown, and returns ObserverNotReachableException for requests to missing observers.
• Expands integration tests for enumeration, batching, cancellation, cleanup, and observer lifecycle behavior.

File	Description
test/​Orleans.DefaultCluster.Tests/​Orleans.DefaultCluster.Tests.csproj	Adds an async-enumerable test folder entry.
test/​Orleans.DefaultCluster.Tests/​ObserverTests.cs	Tests calls to deleted observers.
test/​Orleans.DefaultCluster.Tests/​AsyncEnumerableGrainCallTests.cs	Shares test infrastructure and adds observer enumeration tests.
test/​Orleans.Core.Tests/​Runtime/​InvokableObjectManagerTests.cs	Tests missing-observer responses and shutdown registration rejection.
test/​Grains/​TestInternalGrains/​ObservableGrainObserver.cs	Adds an observer implementation for enumeration tests.
test/​Grains/​TestInternalGrains/​ObservableGrain.cs	Makes asynchronous yielding conditional in the test iterator.
test/​Grains/​TestGrainInterfaces/​ISimpleObserverableGrain.cs	Adds a request-response observer interface.
test/​Grains/​TestGrainInterfaces/​IObservableGrainObserver.cs	Defines observer enumeration test methods.
src/​Orleans.Core/​Runtime/​InvokableObjectManager.LocalObjectData.cs	Extracts observer context handling and adds owned-extension disposal.
src/​Orleans.Core/​Runtime/​InvokableObjectManager.cs	Tracks lifecycle cleanup and responds to missing-observer requests.
src/​Orleans.Core/​Runtime/​InvokableObjectManager.AsyncEnumerableExtension.cs	Implements observer enumeration, synchronization, and cleanup.
src/​Orleans.Core.Abstractions/​Exceptions/​OrleansMessageRejectionException.cs	Removes a trailing blank line; no behavioral change.
src/​Orleans.Core.Abstractions/​Exceptions/​ObserverNotAvailableException.cs	Defines ObserverNotReachableException.

_{Review effort: Balanced}

[github-actions]

Code coverage

Metric	Pull request	Current main	Variance
Lines	83.85% (121,596 / 145,009)	83.88% (121,114 / 144,394)	-0.0233 pp
Branches	73.75% (36,383 / 49,334)	73.68% (36,233 / 49,174)	+0.0651 pp

Report-only conclusion: mixed.

The current-main baseline is commit 8e6a406708 and uses the same reviewed coverage matrix.

Coverage combines every CI test matrix job, including providers, CodeGen, .NET 8/10, Linux, Windows, and macOS, using canonical physical source and branch identities.

The comparison remains report-only while normal line and branch variance is calibrated.

Coverage details

• Pull request commit: aa14c3e4a3
• View the pull request test run and coverage artifacts
• View the trusted coverage reporting run

[ReubenBond]

🤖 Automated review. This comment was generated by an AI-assisted design review: five parallel lens agents (architecture, invariants, testing, performance, code craft), with the results merged and deduplicated. It reviews head aa14c3e4a3 against main (bdbe81dc72). Findings are ordered by priority: correctness > simplicity > performance > convenience.

Summary

The core enumerator protocol in InvokableObjectManager.AsyncEnumerableExtension.cs holds up. No lost or duplicated elements were found, every gate acquisition is released, lock ordering is consistent, and the reference-equality fence prevents a request from continuing on a newer enumerator state. The heartbeat lost-element bug and the dispose-acknowledged-at-cancellation issue from the previous round are fixed in 9841c5a620. CI is green.

The remaining problems are in lifecycle and failure semantics:

• Shutdown waits for in-flight long-polls before it cancels any producer.
• The teardown worker has no owner-bounded lifetime.
• The cleanup loop captures the first caller's execution context.
• Three paths still deviate from the new "missing observer → immediate ObserverNotReachableException" contract.

On the testing side, none of the time- or concurrency-dependent guarantees is pinned by a test: the heartbeat boundary, shutdown with a parked enumerator, and dispose-joins-teardown.

Findings

#	Sev	Lens	Location	Finding	Suggested fix
1	P2	Arch, Inv, Test	InvokableObjectManager.cs StopAsync (~L128–160)	StopAsync awaits _invocations.CloseAsync() before disposing observers. An idle infinite observer stream (the typical case) holds an admission until its long-poll heartbeat, up to ConfiguredResponseTimeout/2 (15 s by default), and on a fake clock indefinitely. Producers are cancelled only after that drain, and an in-flight DisposeAsync RPC can add another ResponseTimeout. Shutdown should seal, then cancel, then join.	Close the gate synchronously (var drain = _invocations.CloseAsync();), dispose or cancel all observers, then await the drain and the disposals together. Also link the long-poll WaitAsync to state.RemovalTokenSource. On removal it returns MissingEnumeratorError and keeps MoveNextTask, so teardown still waits for the producer.
2	P2	Arch, Inv	AsyncEnumerableExtension.cs WaitUntilCompletedAsync (~L701–715), extension DisposeAsync (~L813)	A producer that ignores cancellation keeps its teardown worker looping forever. Each loop logs "I will try again!" every ResponseTimeout and keeps the state, the enumerator and the observer alive. The worker keeps running after extension dispose and after the host's DI scope is disposed. The shutdown deadline is the extension's private ResponseTimeout rather than the host's token. The in-code comment ("maybe a ceiling like 3?!") shows this was never decided. The grain-side extension abandons the producer after its timeout and skips DisposeAsync.	Thread a CancellationToken through StopAsync → LocalObjectData.DisposeAsync → extension DisposeAsync, so the host owns the deadline. When the deadline fires, abandon: don't call DisposeAsync while MoveNextAsync is pending, log once, release the gate, complete teardown and remove the state. Delete the open-ended retry loop.
3	P2	Inv	AsyncEnumerableExtension.cs ctor (L55), StartDisposal called from the cleanup loop	The extension is constructed inside the first StartEnumeration, after RequestContextExtensions.Import. RunCleanupLoopAsync captures that ExecutionContext and restores it on every tick for the extension's lifetime. As a result, producer finally blocks and outbound calls from evicted enumerators run under an unrelated caller's RequestContext (tenant or auth data) and Activity, and that context is retained.	Start the loop under ExecutionContext.SuppressFlow() (and the same when the cleanup loop starts disposal), or clear RequestContext there.
4	P2	Arch, Inv, Test	InvokableObjectManager.Dispatch (~L186–203), LocalObjectData.cs (~L140–147, L81)	The new immediate-rejection contract has three gaps: (a) After StopAsync empties localObjects, requests get ObserverNotReachableException (permanent) instead of the shipped SiloUnavailableException (transient, "host shutting down"). Dispatch_RejectsCancellationControlAfterDrain was changed to match the code. (b) The first request to a GC'd observer is still dropped silently, so that caller waits the full ResponseTimeout. (c) Queued enumerator calls after LocalObjectData disposal get ObjectDisposedException from GetComponent.	(a) Check _isStopping first in Dispatch and route to RejectMessage, then restore the original test assertion. (b) For Direction == Request, send Response.FromException(new ObserverNotReachableException(...)) before disposing. (c) Map this case to MissingEnumeratorError or ObserverNotReachableException.
5	P2	Arch, Craft	src/api/Orleans.Core.Abstractions/Orleans.Core.Abstractions.cs	The generated API surface was not regenerated. It lacks ObserverNotReachableException and its Codec_/Copier_ types; I confirmed this locally by regenerating.	Regenerate it per AGENTS.md ("Generated API surfaces") and commit the result.
6	P3	Inv	AsyncEnumerableExtension.cs ~L422–428	The bare catch after the long-poll WaitAsync assumes the task has completed. If WaitAsync throws synchronously (for example ArgumentOutOfRange for a very large or infinite ConfiguredResponseTimeout), release builds clear a pending MoveNextTask. Teardown can then call DisposeAsync while MoveNextAsync is still running. The comments on the TimeoutException and bare catch also describe the wrong exceptions: a cancelled inner task surfaces as TaskCanceledException.	Use await moveNextTask.WaitAsync(...).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing) and then classify. This removes both catch blocks.
7	P3	Inv	AsyncEnumerableExtension.cs L185	MaxBatchSize comes from the remote caller but is checked only by Debug.Assert.	Return (Error, ArgumentOutOfRangeException) when it is ≤ 0.
8	P3	Inv, Arch	AsyncEnumerableExtension.cs ~L514, terminating MoveNext path	Server-side waits are at least as long as the client's timeout. The DisposeAsync RPC waits exactly ResponseTimeout, so a slow teardown shows up as a caller timeout rather than a clean reply. The RPC cancellationToken is checked only on entry.	Bound server waits strictly below the caller's timeout (for example ConfiguredResponseTimeout/2, like the long-poll) and pass the token through.
9	P3	Arch, Inv, Craft	AsyncEnumerableExtension.cs ~L813	Task.WhenAll(teardownTasks).WaitAsync(_messagingOptions.ResponseTimeout) omits _timeProvider. Every other wait uses it.	Pass _timeProvider. This becomes moot if finding 2 adopts a host-owned deadline.
10	P1 (test)	Test, Inv	AsyncEnumerableGrainCallTests.cs	Key guarantees have no test: • Ordering with no loss or duplication across a heartbeat. The client long-poll is never forced to a heartbeat, so the fix in 9841c5a620 is unverified. • StopAsync with a parked enumerator. • That explicit dispose joins producer teardown. EnumeratorDisposed_WhileMoveNextIsPending would also pass if dispose returned early.	Heartbeat: a deterministic test with a TCS-driven producer and a barrier until it is parked. Advance the client clock by CRT/2, wait for the second pull, release one value, and assert the exact sequence and pull count. Add variants that cancel inside the heartbeat window and that race a value with the heartbeat. StopAsync: a unit test in InvokableObjectManagerTests asserting completion without advancing time. Dispose: gate the producer's finally, assert DisposeAsync is still pending, then release the gate.
11	P1 (test)	Test	AsyncEnumerableGrainObserverExtension	The extension's lock-plus-gate state machine has no unit or model-based tests. Uncovered properties: • concurrent MoveNext on one ID is serialized • a duplicate ID is rejected • unknown, disposing or evicted IDs return MissingEnumeratorError • a wrong T returns InvalidCastException • StartEnumeration after dispose returns ObjectDisposedException • the cleanup loop skips gated (in-flight) enumerators • producer DisposeAsync never overlaps MoveNextAsync • Seen expiry takes exactly two ticks	Construct the extension directly in Orleans.Core.Tests with a fake IGrainContext, a hand-built IAsyncEnumerableRequest<T> and a FakeTimeProvider. Add an Accordant model-based test over Start, MoveNext, Dispose, Tick, DisposeExtension and producer Yield/Complete/Fault/Hang.
12	P2 (test)	Test	AsyncEnumerableGrainCallTests.cs fixture (~L45)	Replacing TimeProvider on the client without UseTimeProviderForBackgroundAreas(TimeProvider.System) also moves client messaging callbacks and the gateway manager onto the fake clock (see test/AGENTS.md). A hung call then runs until the xUnit timeout instead of failing with a diagnosis. AdvanceTimeByResponseTimeoutAsync drives the client extension using the silo's ResponseTimeout.	Add clientBuilder.Services.UseTimeProviderForBackgroundAreas(TimeProvider.System), and derive the advance amount from the client's MessagingOptions.
13	P2 (test)	Test, Craft	AsyncEnumerableGrainCallTests.cs (~L729, ~L1370, ~L1436, ~L1496, ~L1694), ObserverTests.cs (~L387–422)	Test hygiene problems: • Both SlowProducer tests now await producer after breaking at item 2, adding ~6 s of idle time each to BVT. They don't exercise heartbeats either, since 2 s is far below CRT/2 and the long-poll runs on the fake clock. • The producer in WithCancellation escapes the test. • catch (Exception) { } and Task.WhenAny(singleTask) make assertions unable to fail. • ObserverTest_CallDeletedObserver uses an untracked Task.Run polling loop.	• Cancel producers through a linked CTS after break, then await them. • Assert the allowed exception set (OperationCanceledException, EnumerationAbortedException, ObserverNotReachableException). • Replace the polling loop with await Assert.ThrowsAsync<ObserverNotReachableException>(() => observerRef.Ping()), since DeleteObjectReference deregisters synchronously.
14	P2	Arch, Craft	AsyncEnumerableExtension.cs ~L321–470 compared with AsyncEnumerableGrainExtension.cs ~L166–300	The batch-drain loop, long-poll classification and termination mapping are copied from the grain extension, and the copies already disagree: • Wrong element type: the grain side throws and keeps the enumerator; the observer side returns Error and tears it down. • Teardown timeout: the grain side abandons; the observer side retries forever. • Idle sweep: the grain side marks busy enumerators; the observer side skips any gated one.	Extract the pure protocol steps into shared internal static helpers: TryPullBatch<T>, ClassifyCompletedMoveNext<T>, and IsAbnormalTermination() next to IsTerminated(). Keep only the concurrency layer separate. This also removes most of the 11 goto Done jumps in MoveNextCoreAsync, because the synchronous parts don't need to live inside the async body.
15	P2	Perf, Arch, Inv	AsyncEnumerableExtension.cs L55/L86	Once an observer has served any enumeration, its PeriodicTimer cleanup loop runs until the observer is deregistered. Observers aren't deactivated when idle, so 10k such observers mean ~10 MB retained and ~333 ticks/s at the default 30 s ResponseTimeout. GC'd observers that are never deleted, an existing leak, now also keep a ticking timer.	Start the loop when _enumerators goes from 0 to 1, and let it exit when a tick finds the dictionary empty. Alternatively, use one manager-level sweep over extensions with live enumerators.
16	P3	Perf	AsyncEnumerableExtension.cs ~L469/589/683, LocalObjectData.cs ~L23–31/598, InvokableObjectManager.cs ~L61–65	Smaller costs: • Normal Completed still waits for the disposal acknowledgement before replying: ~1–1.5 KB, three thread-pool hops and a timer per enumeration. • new LocalObjectData(...) now runs under the global lifecycle lock. • DisposeAsync allocates a TCS even when there is no extension. • The dedicated per-observer Lock adds ~56 B per observer that never enumerates.	• Skip the acknowledgement wait for normal completion. • Allocate before taking the lock. • Use a sentinel when there is no extension. • Create the extension with Interlocked.CompareExchange and drop the dedicated lock.
17	P3	Craft	AsyncEnumerableExtension.cs ~L518–550, ~L552–562, ~L838–880	Naming and dead code: • RequestDisposal's expectedState is null branch is unreachable (every caller passes a state), and requestId is unused. • DisposalTask completes when cancellation is signalled, before disposal. • GetOrCreateDisposalTask also cancels RemovalTokenSource. • The rationale for the void/out signature ("so nobody awaits it") doesn't hold, because C# already forbids await in lock.	• Change the signature to RequestDisposal(EnumeratorState state, bool cancel). • Rename to CancellationSignaled/TeardownCompleted and BeginDisposal(...). • Drop the misleading comment.
18	P3	Craft	various	Hygiene: • ObserverNotAvailableException.cs defines ObserverNotReachableException; rename the file. "not longer" should be "no longer". • using Microsoft.CodeAnalysis; is unused in AsyncEnumerableExtension.cs and LocalObjectData.cs. • The trailing-newline removal in OrleansMessageRejectionException.cs is unrelated. • Lock uses #if NET10_0_OR_GREATER, whereas the repo convention is NET9_0_OR_GREATER (see ClientGrainContext.cs). • Typos: "dispoal", "immedialty", "completation", "consuder", "reentracy", "TryTryDeregister", "implemention", "thread-saftey". There are also concatenated justification strings missing spaces (InvokableObjectManager.cs ~L70–72). • [TestSuite]/[TestProvider] are duplicated on the class and every method. • ThrowsAnyAsync<ObserverNotReachableException> on a sealed type should be ThrowsAsync.	Mechanical cleanup.

Adjacent, pre-existing on main (not introduced by this PR): ProcessMessageAsync calls message.Dispose() on expired messages, which nulls the body. The finally block then calls GetAdmissionGate(message) again, and IsCancellationRequest reads the now-null body, so an expired IGrainCallCancellationExtension message exits _invocations instead of _cancellations. As a result, _invocations underflows and starts rejecting observer calls, and _cancellations leaks an admission, so StopAsync waits until the host deadline. The fix is to capture the gate at admission and pass it through. This is worth a separate fix and issue.

Questions for the maintainer and author

1. Producer that never completes: should teardown wait forever (current behavior) or abandon the producer after a host-owned deadline, as the grain-side extension does? This decides finding 2.
2. Shutdown error type: after StopAsync, should callers get SiloUnavailableException (transient, shipped behavior) or ObserverNotReachableException (permanent)? Observer-manager patterns may permanently unsubscribe on the latter. See finding 4a.
3. Exception base class: should ObserverNotReachableException derive from OrleansMessageRejectionException, so existing catch handlers keep working, or stay on OrleansException? This cannot change compatibly after it ships.
4. Rolling upgrade: an old caller cannot resolve the new exception type, so it receives the deserializer's fallback exception immediately instead of a TimeoutException; is that acceptable? A new caller enumerating on an old observer host falls through GetComponent to rootGrainContext. Should that be a defined, typed error and documented as "the observer host must be upgraded"?
5. Client restart: StopAsync now permanently removes every observer registration, whereas registrations previously survived it. Is restarting an OutsideRuntimeClient with live observers supported?
6. Observer lifetime: an active or abandoned enumeration holds the observer strongly, through the enumerator closure, for up to about 2×ResponseTimeout, even though the registration is weak. Should the observer docs state this?
7. Cancel under lock: is "the only registrants on RemovalTokenSource are SemaphoreSlim waits, whose continuations run asynchronously" an intended invariant? If so, document it at GetOrCreateDisposalTask, because any future registration would run under _lock.