Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 47 additions & 7 deletions src/MeshWeaver.AI.Orleans.TestBase/SharedOrleansFixture.cs
Original file line number Diff line number Diff line change
Expand Up @@ -189,20 +189,60 @@ public async Task CleanupClientAsync(IMessageHub client)
catch { /* tearing-down — swallow so other cleanups still run */ }
}

// Captured while the hub's scope is still alive — a late fault has to land SOMEWHERE, and
// "somewhere" cannot be a service resolved after disposal began (the same rule
// MeshTeardownExtensions states for every teardown service).
var logger = SafeLogger(reg.Hub);

try { reg.Hub.Dispose(); }
catch { /* same */ }
if (reg.Hub.IsDisposing)
{
// Bridge reactive completion → Task at this teardown edge; swallow faults/timeout.
// 🚨 SUBSCRIBE; bound the WAIT with a token, never a Timeout spliced into the signal
// (#2301/#2488). The old shape — `Catch(…).FirstOrDefaultAsync().ToTask().WaitAsync(10s)`
// inside a bare `catch { }` — resumed this cleanup INLINE on the client hub's own
// disposal thread (Rx completes a ToTask() TCS without RunContinuationsAsynchronously),
// and then discarded three different outcomes as one silence: a disposal fault, a fault
// arriving after the bridge settled, and the budget expiring. Cleanup still proceeds on
// all three (a test fixture must not hang on a client hub), but it no longer rides a hub
// thread to get there, and each outcome is now SAID.
using var disposalBudget = new CancellationTokenSource(TimeSpan.FromSeconds(10));
try
{
await reg.Hub.DisposalCompleted
.Catch<Unit, Exception>(_ => Observable.Return(Unit.Default))
.FirstOrDefaultAsync()
.ToTask()
.WaitAsync(TimeSpan.FromSeconds(10));
await reg.Hub.DisposalCompleted.ObserveCompletion(
ex => logger?.LogError(ex,
"Client hub {Address}: disposal faulted AFTER cleanup stopped waiting on it "
Comment thread
rbuergi marked this conversation as resolved.
+ "— reported rather than orphaned.", reg.Hub.Address),
disposalBudget.Token);
}
catch (OperationCanceledException)
{
logger?.LogError(
"Client hub {Address}: disposal did not complete within 10s during test cleanup. "
+ "Something on its action block is not finishing.", reg.Hub.Address);
}
catch { /* timeout / already-faulted — don't block teardown */ }
catch (Exception ex)
{
logger?.LogError(ex, "Client hub {Address}: disposal FAULTED during test cleanup.",
reg.Hub.Address);
}
}
}

/// <summary>
/// A logger from a hub whose scope may be mid-teardown — resolved defensively so cleanup can
/// never fail on the act of preparing to REPORT something.
/// </summary>
private static ILogger? SafeLogger(IMessageHub hub)
{
try
{
return hub.ServiceProvider.GetService<ILoggerFactory>()?
.CreateLogger("MeshWeaver.Hosting.Orleans.Test.Cleanup");
}
catch
{
return null;
}
}

Expand Down
29 changes: 27 additions & 2 deletions src/MeshWeaver.Connection.Orleans/IoPoolSiloTeardown.cs
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,33 @@ Task ILifecycleObserver.OnStart(CancellationToken cancellationToken)

// 🚨 NOT `async`, and nothing here awaits. AGENTS.md: everything is IObservable<T> end-to-end —
// compose and Subscribe, never await. The single Task exists only because ILifecycleObserver
// demands one, and it is produced by ONE .ToTask() at the boundary — the sanctioned shape for a
// framework surface whose body stays reactive.
// demands one, and it is produced by ONE .ToTask() at the boundary.
//
// 🚨 THAT Task IS GENUINELY REQUIRED, and it was re-verified rather than inherited from the
// IMessageHub.DisposalCompleted documentation that turned out to be wrong (#2301). Orleans'
// Orleans.ILifecycleObserver declares `Task OnStop(CancellationToken)`; the silo AWAITS what it
// returns, and that await IS the guarantee that the silo does not release over live pool
// threads. There is no observable-shaped overload to move to.
//
// 🚨 THE DEADLOCK QUESTION FIRST — "whose thread am I blocking, and does the signal I am
// waiting for need it?" Here: NOBODY'S. This method composes and RETURNS; it never awaits, so
// no scheduler is held while registry.Disposed is pending. The bridge sits at the OUTERMOST
// edge, which is the whole rule: a .ToTask() that is HANDED to the API demanding it is safe,
// a .ToTask() that you then `await` in the middle of your own work is how #2301 parked the
// grain scheduler its next wait needed. (/async skill, Rule 1a: subscribe, hand back a Task,
// do not await inside the turn.)
//
// The two secondary properties, both checked:
// • every fault is observed BEFORE the bridge — the .Catch below is inside the composition,
// so nothing can fault past .ToTask() into an unobserved task;
// • IoPoolRegistry.Disposed is an AsyncSubject<int>: it fires ONCE and replays. It cannot
// emit a value and then fault, so there is no "late fault after the interesting value"
// for the settled Task to miss. (Where a signal CAN do that — anything composed, merged,
// or polled — use ReactiveCompletion.ObserveCompletion, which keeps its error arm
// attached after the task settles AND completes with RunContinuationsAsynchronously.)
// • the .Timeout is a REPORTED budget on a join, not a silent race: expiry lands in Report(-1)
// as a LogError naming the consequence. It is the weaker class #2488 lists at 4–5, kept
// because a silo that never releases is worse than one that releases loudly.
//
// Orleans awaits the returned Task, which is what holds the silo back until the pools report.
// That is the whole guarantee, and it costs no thread: nothing is parked here, the completion
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,91 @@ var nodePaths = level.NodePaths.ToInlineObservable();

---

## 🚨 Waiting for a completion signal: `Subscribe`, never `.ToTask()`

Disposal, teardown, a drain report, a grain deactivation — every "wait for X to finish" in the mesh
is an `IObservable<T>` that fires once and terminates. **Waiting for one means
`signal.Subscribe(onNext, onError)` and nothing else.** No poll, no `SpinWait`, no
`.Wait()`/`.Result`, no `Timeout` operator raced against the signal, and no `Task` gate.

### The reason is DEADLOCK, not a lost exception

**These completions are produced BY THE VERY SCHEDULER the waiter is occupying.** A hub's disposal
finishes when its single-threaded action block drains; a grain's deactivation finishes when its turn
scheduler is free. Wait on one of those from that same scheduler and the thing you are waiting for
can never happen — the Task never settles, and the only thing that ever ends the wait is whatever
timeout was raced against it. This is [Rule 1 of the `/async` skill](/Doc/Architecture/AsynchronousCalls)
applied to lifecycle signals.

🚨 **You do not have to write the block to get it — Rx hands it to you.** `ToTask()` completes its
`TaskCompletionSource` from inside the pipeline **without `RunContinuationsAsynchronously`**, so
`TrySetResult` resumes the awaiter **inline on the signalling thread**. Everything after that
`await` — the rest of your method — then runs on the hub's disposal thread or the grain's turn
scheduler. And it is sticky: with no `SynchronizationContext`, `await` captures
`TaskScheduler.Current`, so **one** inline resumption routes **every later await in that method**
onto the same scheduler.

That is issue #2301. `OrleansGrainTeardownStragglerTest` resumed inline on the deactivating grain's
scheduler and then held it, waiting for that grain's activation to leave the silo catalog — which
needed the scheduler it was holding. It failed at **exactly 30 s**, its `Timeout` budget, every
time, while a healthy activation leaves the catalog in **0.10 s**. A number that is always the
budget rather than a distribution around it is the signature of a deadlock, not of contention.

### The lost fault is the second-order effect

When that timeout finally fires it settles the Task, and a fault still travelling the chain has **no
observer left** — an unobserved exception, surfaced on the finalizer as
`UnobservedTaskException`, which xUnit v3 escalates into a Catastrophic failure that poisons the
*next* test class. That is #2301's `HOST_CRASHED` marker: what the deadlock does on its way out.

`IMessageHub.DisposalCompleted`'s own documentation used to recommend the broken bridge (*"at a
genuine async edge (test teardown, grain deactivation) bridge once with
`DisposalCompleted.FirstOrDefaultAsync()` / `.ToTask()`"*), which is why four attempted fixes all
aimed at the timeout. That sentence is gone.

```csharp
// ❌ WRONG — resumes the caller INLINE on the hub's disposal thread, then holds it; and the
// Catch DISCARDS a faulted disposal on top.
await hub.DisposalCompleted
.Catch<Unit, Exception>(_ => Observable.Return(Unit.Default))
.FirstOrDefaultAsync()
.ToTask()
.WaitAsync(timeout); // …racing what it waits for

// ✅ RIGHT — the turn returns immediately and the follow-on work belongs to the signal. The error
// arm is fluent (.Catch) because a fault arrives LATER and on ANOTHER thread: a try/catch
// around the Subscribe CALL cannot see it. This is MessageHubGrain.OnDeactivateAsync.
hub.DisposalCompleted
.Take(1)
.Catch<Unit, Exception>(ex => { logger.LogError(ex, "… KEEPING its load context"); return Observable.Empty<Unit>(); })
.Subscribe(_ => UnloadContextIfSafe(reason, grainId));
return Task.CompletedTask;
```

### Where an API genuinely demands a `Task`

`ILifecycleObserver.OnStop`, `IHostedService`, an `async Task` test method. Bridge with
`ReactiveCompletion.ObserveCompletion`, which subscribes, completes its task with
`RunContinuationsAsynchronously` — so the caller is **never** resumed on the signalling scheduler —
and keeps its error arm attached after the task settles. Two rules go with it:

- **Put the bridge at the OUTERMOST edge.** Return the Task to whoever demanded it; never wrap it
around a wait whose completion needs the thread you are holding. `IoPoolSiloTeardown.OnStop` is
the reference: it composes and **returns**, blocking nothing.
- **The bound belongs to the edge, not to the signal.** `[Fact(Timeout = …)]`, a host's shutdown
token, a `CancellationToken` — never a `Timeout` operator spliced into the wait, which settles the
waiter while the work it bounded is still in flight.
- **A poll is the same defect wearing a loop.** If you are sampling state on an interval because
there is no completion signal to subscribe to, publish the signal. #2301's test polled
`IManagementGrain.GetDetailedGrainStatistics()` every 100 ms against a 30 s `Timeout` because
nothing exposed "this activation is fully gone"; the fix was `GrainDeactivationCompleted`, which
the grain publishes from Orleans' own `IGrainContext.Deactivated`.

Both properties are pinned by `DisposalWaitBridgeTest` (test/MeshWeaver.Messaging.Hub.Test), whose
control test asserts that `.ToTask()` really does resume inline on the signalling thread.

Related: issue #2488 catalogues the remaining poll / timeout-escape sites.

## 🚨 Cold observables: Subscribe is mandatory

Every method that performs a write or side effect returns a cold `IObservable<T>` — **the side effect runs on `Subscribe`, not on call.** Forgetting to subscribe means the work silently never happens.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
---
Name: Shutdown says what went wrong
Category: Fix
Description: A fault while a mesh shuts down is now reported instead of being silently discarded.
Icon: Sparkle
Order: -20260827
---

# Shutdown says what went wrong

When a mesh shuts down, the code that waits for it to finish used to convert the wait into a
one-shot handle that could only ever report one thing. Anything that went wrong after that
handle had already settled — a hub whose teardown failed a moment later, a background job that
faulted on its way to idle — reached nobody at all: no log line, no error, nothing to search for
afterwards.

Every one of those waits now stays subscribed for as long as the shutdown can still say
something, so a late failure is written to the log with the address it came from. The shutdown
report also carries two facts it previously threw away: whether the shutdown itself faulted, and
whether running activities actually finished before the mesh was torn down.

Nothing about a healthy shutdown changes. What changes is that an unhealthy one is now visible
instead of looking identical to a clean one.
80 changes: 58 additions & 22 deletions src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs
Original file line number Diff line number Diff line change
Expand Up @@ -1613,36 +1613,72 @@ private void WriteInstanceMemoryDelta(string testName, bool shared)
}

/// <summary>
/// Awaits <see cref="IMessageHub.DisposalCompleted"/> with periodic progress snapshots.
/// Waits for <see cref="IMessageHub.DisposalCompleted"/> with periodic progress snapshots.
/// Every <see cref="DisposeProgressInterval"/>, dumps
/// <see cref="IMessageHub.GetDisposalDiagnostics"/> to <see cref="MeshWeaver.Fixture.TestBase.FileOutput"/>
/// so a hang shows up as a stream of snapshots converging on the offending hub
/// — instead of one big snapshot at the timeout.
///
/// <para>🚨 <b>Both halves are SUBSCRIPTIONS now (#2301, #2488 site 3).</b> This used to bridge
/// the reactive signal to a Task with <c>Catch(…).FirstOrDefaultAsync().ToTask()</c> and then
/// POLL that Task in a <c>while (!disposal.IsCompleted)</c> loop — the canonical
/// "wait for disposal" every test in the suite runs, so it shaped what teardown failures look
/// like everywhere.</para>
///
/// <para>The first defect was DEADLOCK, not the discarded fault. Rx completes a
/// <c>ToTask()</c> <c>TaskCompletionSource</c> from inside the pipeline WITHOUT
/// <c>RunContinuationsAsynchronously</c>, so <c>DisposeAsync</c> resumed INLINE on the thread
/// that signalled disposal — the mesh hub's own — and then ran the rest of teardown there,
/// including <c>ioPools.DrainAll()</c>, a SYNCHRONOUS JOIN of every pooled I/O leaf. The
/// second was that the <c>Catch</c> discarded a disposal fault outright, and the Task could
/// observe nothing once it settled. The progress ticker is now an <c>Observable.Interval</c>
/// that stops on the signal's own terminal, and the wait is one <c>Subscribe</c> whose task
/// completes asynchronously, with an error arm.</para>
/// </summary>
private async Task WaitWithProgressAsync(string testName, Stopwatch sw, CancellationToken ct)
{
// Bridge the reactive completion to a Task once, at this test-teardown edge. Catch folds
// a disposal fault into completion (teardown only waits for "done", not why).
var disposal = Mesh.DisposalCompleted
.Catch<Unit, Exception>(_ => Observable.Return(Unit.Default))
.FirstOrDefaultAsync()
.ToTask();
while (!disposal.IsCompleted)
var disposal = Mesh.DisposalCompleted;

// Progress ticker: stops on the signal's terminal, whatever it is. Materialize() so a
// FAULTED disposal ends the ticker as data rather than pushing an error into a
// subscription that has no arm for it.
using var progress = Observable.Interval(DisposeProgressInterval)
.TakeUntil(disposal.Materialize())
.Subscribe(
_ =>
{
FileOutput.WriteLine(
$"[DISPOSE] {testName}: still waiting after {sw.ElapsedMilliseconds}ms — snapshot:");
FileOutput.WriteLine(SafeGetDiagnostics());
},
ex => FileOutput.WriteLine(
$"[DISPOSE] {testName}: progress ticker faulted: {ex.GetType().Name}: {ex.Message}"));

try
{
using var progressCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
progressCts.CancelAfter(DisposeProgressInterval);
try
{
await disposal.WaitAsync(progressCts.Token);
return; // disposal completed
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
// Progress tick fired before disposal completed — log a snapshot and loop.
FileOutput.WriteLine(
$"[DISPOSE] {testName}: still waiting after {sw.ElapsedMilliseconds}ms — snapshot:");
FileOutput.WriteLine(SafeGetDiagnostics());
}
await disposal.ObserveCompletion(
ex => FileOutput.WriteLine(
$"[DISPOSE] {testName}: disposal faulted AFTER the wait settled — reported, "
Comment thread
rbuergi marked this conversation as resolved.
+ $"not orphaned: {ex.GetType().Name}: {ex.Message}"),
ct);
}
catch (OperationCanceledException)
{
// The hang path — unchanged. DisposeAsync turns this into a loud TimeoutException
// carrying the hub diagnostics.
throw;
}
catch (Exception ex)
{
// A disposal FAULT arriving as the answer. Teardown still waits for "done", not "why"
// — but the fault is now REPORTED into the per-test output and the phase trace instead
// of being discarded by a .Catch nobody could see. (Whether it should also FAIL the
// class is an escalation decision of its own; see TeardownReport.DisposalFault.)
FileOutput.WriteLine(
$"[DISPOSE] {testName}: disposal FAULTED after {sw.ElapsedMilliseconds}ms: "
+ $"{ex.GetType().Name}: {ex.Message}");
TestPhaseTrace(testName, "DISPOSE_FAULTED", sw.ElapsedMilliseconds,
$"{ex.GetType().Name}: {ex.Message}");
}
}

Expand Down
Loading
Loading