From 3ef56be2dd2d707b306404f714f013946eacac53 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 21:21:25 -0400 Subject: [PATCH] fix(worker): unconditional fault observation for abandoned pipe I/O; exception-total transport dispose --- docs/MxAccessWorkerInstanceDesign.md | 42 ++- .../Ipc/WorkerPipeSessionTests.cs | 239 ++++++++++-------- .../Ipc/WorkerPipeSession.cs | 126 +++++++-- 3 files changed, 262 insertions(+), 145 deletions(-) diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 35c6cf1..28f49a1 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -900,22 +900,36 @@ heartbeat and event-drain loops, so every fault exit — an event-drain fault, a event too large to frame, a failed heartbeat write — unwinds while that read is still pending. The session therefore owns the transport: `RunAsync`'s outermost `finally` disposes the stream as its last teardown step and then awaits the read -that disposal unblocks. Both halves matter. Disposal has to come last because -every frame the session will ever write is complete by then (the frame writer -signals a write only after it has been written *and* flushed), and the await has -to happen while the session still holds the read task, because the worker -installs no `TaskScheduler.UnobservedTaskException` handler — otherwise the -read's `ObjectDisposedException`/`IOException` lands on a task nobody observes, -still holding the reader's reused length-prefix buffer and its pooled payload -buffer. +that disposal unblocks. Disposal comes last because in the ordinary case every +frame the session will ever write is already complete by then — the frame writer +signals a write only after it has been written *and* flushed. It is not last +because that is guaranteed: the wait on the heartbeat and drain loops is +budgeted, and a stream write is genuinely uncancellable, so an overrunning write +can still be in flight against the stream being disposed. Disposal is +consequently exception-*total*, catching anything the handle close throws and +logging it, because nothing raised while releasing a handle is more actionable +than the terminal exception that ended the session, and nothing may displace it. + +Observation is unconditional; only the *logging* of it is budgeted. +`ObserveBackgroundTaskStopAsync` waits `BackgroundTaskStopTimeout` for a task to +stop and logs what it saw, but when it gives up it hands the task a +fault-observing continuation before returning. Windows owes no deadline for a +completion torn off a closed handle, so a bounded await on its own would reopen +the very orphaning window it was added to close. The same helper — and so the +same guarantee — covers the abandoned read, the heartbeat loop, and the +event-drain loop. This matters because the worker installs no +`TaskScheduler.UnobservedTaskException` handler: an unheld faulted task would +otherwise surface only at finalization, still holding the reader's reused +length-prefix buffer and its pooled payload buffer. Two invariants follow. Nothing may call `WorkerFrameReader.ReadAsync` again once -a read has been abandoned: a second read would race the first for those buffers -and could return a pooled buffer twice. And `WorkerPipeClient`'s `using` on the -pipe stays as a backstop for the paths the session never reaches (a session -factory that throws), not as the primary owner — disposal is idempotent, so its -second `Dispose` is a no-op. Graceful shutdown leaves no pending read at all, so -the observation step is a no-op on that path. +a read has been abandoned — a second read would race the first for those buffers +and could return a pooled buffer twice — which `Debug.Assert`s at both read-issue +sites guard. And `WorkerPipeClient`'s `using` on the pipe stays as a backstop for +the paths the session never reaches (a session factory that throws), not as the +primary owner; disposal is idempotent, so its second `Dispose` is a no-op. +Graceful shutdown leaves no pending read at all, so the observation step is a +no-op on that path. `MxAccessStaSession.ShutdownGracefullyAsync` implements the current cleanup path. It first calls `StaCommandDispatcher.RequestShutdown()` so new commands diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 18580cd..826e1b0 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -912,112 +912,6 @@ public sealed class WorkerPipeSessionTests await Assert.ThrowsAsync(async () => await runTask); } - /// - /// WRK-31. A fault exit unwinds the message loop while its frame read is still pending, and - /// on net48 nothing can cancel that read — NamedPipeClientStream.ReadAsync ignores - /// the token, so only closing the handle ends it. The session must therefore dispose the - /// transport itself and then await the read that disposal unblocks: the worker installs no - /// TaskScheduler.UnobservedTaskException handler, so before this the read faulted on - /// a task nobody held — carrying the reader's reused prefix buffer and its pooled payload - /// buffer with it — and surfaced only when the finalizer got round to it. - /// - /// A task that represents the asynchronous operation. - [Fact] - public async Task RunAsync_WhenFaultEndsSession_ObservesTheAbandonedPipeRead() - { - const uint tinyMaxFrameBytes = 4096; - object unobservedGate = new(); - List unobservedPipeExceptions = new(); - EventHandler unobservedHandler = (_, args) => - { - // TaskScheduler.UnobservedTaskException is process-global and xUnit runs this - // assembly's test classes in parallel, so the capture is narrowed to the failure under - // test: a pipe stream's own teardown exception. SetObserved is deliberately NOT called - // — the default policy already swallows these, and observing them here would mask the - // very regression a concurrently running test might be reporting. - foreach (Exception inner in args.Exception.Flatten().InnerExceptions) - { - if ((inner is ObjectDisposedException || inner is IOException) - && inner.Message.IndexOf("pipe", StringComparison.OrdinalIgnoreCase) >= 0) - { - lock (unobservedGate) - { - unobservedPipeExceptions.Add(inner); - } - } - } - }; - - RecordingWorkerLogger logger = new(); - TaskScheduler.UnobservedTaskException += unobservedHandler; - try - { - using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); - using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); - FakeRuntimeSession runtime = new(); - WorkerPipeSession session = CreatePipeSession( - pipePair.WorkerStream, - runtime, - new WorkerPipeSessionOptions - { - HeartbeatInterval = TimeSpan.FromMilliseconds(100), - HeartbeatGrace = TimeSpan.FromSeconds(5), - }, - logger); - runtime.EnqueueEvent(CreateOversizedWorkerEvent(sequence: 91, payloadBytes: 16 * 1024)); - Task runTask = session.RunAsync(cancellation.Token); - await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token); - - await ReadUntilAsync( - pipePair.GatewayReader, - WorkerEnvelope.BodyOneofCase.WorkerFault, - cancellation.Token); - - // The same 5s bound the sibling oversized-event test uses: teardown must not stall on - // the read it abandoned. - Task completedTask = await Task.WhenAny( - runTask, - Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token)); - Assert.Same(runTask, completedTask); - await Assert.ThrowsAsync(async () => await runTask); - - // Deterministic evidence the read was both abandoned and observed: the shared - // observe-with-timeout helper logs the fault it swallowed, tagged "PipeRead". An - // assertion on the exception type would be over-specified — a handle closed under a - // pending overlapped read surfaces as ObjectDisposedException, IOException, or a - // zero-byte read mapped to EndOfStream depending on how the I/O completes — and the - // contract here is that the fault is observed at all, not which one it is. - Assert.Contains( - logger.Events, - entry => entry.EventName == "WorkerPipeSessionBackgroundTaskStopFailed" - && entry.Fields.TryGetValue("task", out object? task) - && (task as string) == "PipeRead"); - - // ...and that the disposal is what ended it: had the read stayed parked, the helper - // would have given up after BackgroundTaskStopTimeout and logged the timeout instead. - Assert.DoesNotContain( - logger.Events, - entry => entry.EventName == "WorkerPipeSessionBackgroundTaskStopTimedOut" - && entry.Fields.TryGetValue("task", out object? task) - && (task as string) == "PipeRead"); - - // Drive any task that faulted without an awaiter through its finalizer, which is what - // raises UnobservedTaskException. Nothing from the pipe read may surface. - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); - } - finally - { - TaskScheduler.UnobservedTaskException -= unobservedHandler; - } - - lock (unobservedGate) - { - Assert.Empty(unobservedPipeExceptions); - } - } - /// /// WRK-31, the other side of the invariant. The graceful path leaves the message loop /// through its return after the shutdown ack, with that iteration's read already @@ -2381,6 +2275,123 @@ public sealed class WorkerPipeSessionTests return envelopes.ToArray(); } + /// + /// The one teardown test that has to arm , + /// which is process-global: a task faulting in any concurrently running test class can be + /// finalized inside this test's window and read as its result. It therefore lives in its own + /// non-parallel collection (see ) rather + /// than alongside its siblings. Nested so it can still reach + /// 's private harness — PipePair, + /// CreatePipeSession, RecordingWorkerLogger — without widening any of it. + /// + [Collection(WorkerPipeSessionNonParallelCollection.Name)] + public sealed class AbandonedPipeReadTeardownTests + { + /// + /// WRK-31. A fault exit unwinds the message loop while its frame read is still pending, + /// and on net48 nothing can cancel that read — NamedPipeClientStream.ReadAsync + /// ignores the token, so only closing the handle ends it. The session must therefore + /// dispose the transport itself and account for the read that disposal unblocks: the + /// worker installs no TaskScheduler.UnobservedTaskException handler, so before + /// this the read faulted on a task nobody held — still carrying the reader's reused + /// prefix buffer and its pooled payload buffer — and surfaced only at finalization. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_WhenFaultEndsSession_ObservesTheAbandonedPipeRead() + { + const uint tinyMaxFrameBytes = 4096; + object unobservedGate = new(); + List unobservedPipeExceptions = new(); + EventHandler unobservedHandler = (_, args) => + { + // Narrowed to a pipe stream's own teardown exception even though the collection is + // non-parallel, because the handler stays armed across this test's own async + // machinery. SetObserved is deliberately NOT called — the default policy already + // swallows these, and observing them here would mask a regression rather than + // report it. + foreach (Exception inner in args.Exception.Flatten().InnerExceptions) + { + if ((inner is ObjectDisposedException || inner is IOException) + && inner.Message.IndexOf("pipe", StringComparison.OrdinalIgnoreCase) >= 0) + { + lock (unobservedGate) + { + unobservedPipeExceptions.Add(inner); + } + } + } + }; + + RecordingWorkerLogger logger = new(); + TaskScheduler.UnobservedTaskException += unobservedHandler; + try + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(100), + HeartbeatGrace = TimeSpan.FromSeconds(5), + }, + logger); + runtime.EnqueueEvent(CreateOversizedWorkerEvent(sequence: 91, payloadBytes: 16 * 1024)); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token); + + await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerFault, + cancellation.Token); + + // The same 5s bound the sibling oversized-event test uses: teardown must not stall + // on the read it abandoned. + Task completedTask = await Task.WhenAny( + runTask, + Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token)); + Assert.Same(runTask, completedTask); + await Assert.ThrowsAsync(async () => await runTask); + + // Evidence the read was abandoned and that teardown took responsibility for it: the + // shared observe-with-timeout helper records it under the "PipeRead" tag either way. + // Which of the two entries lands is a timing detail, not a contract. The fault + // normally arrives at once (StopFailed), but Windows owes no deadline for a + // completion torn off a closed handle, so on a loaded box it can arrive after + // BackgroundTaskStopTimeout (StopTimedOut). Both are correct, because observation is + // unconditional — the helper attaches a fault-observing continuation when it gives + // up waiting — and the unobserved-exception assertion below is what actually pins + // that. That the transport really is closed is pinned deterministically by + // RunAsync_WhenSessionEnds_ClosesTransportBeforeTheHarnessDoes, so insisting on + // "StopFailed within 1s" here would buy nothing but a flake at the windev gate. + Assert.Contains( + logger.Events, + entry => (entry.EventName == "WorkerPipeSessionBackgroundTaskStopFailed" + || entry.EventName == "WorkerPipeSessionBackgroundTaskStopTimedOut") + && entry.Fields.TryGetValue("task", out object? task) + && (task as string) == "PipeRead"); + + // Drive any task that faulted without an awaiter through its finalizer, which is + // what raises UnobservedTaskException. Nothing from the pipe read may surface. + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + } + finally + { + TaskScheduler.UnobservedTaskException -= unobservedHandler; + } + + lock (unobservedGate) + { + Assert.Empty(unobservedPipeExceptions); + } + } + } + private sealed class RecordingWorkerLogger : ZB.MOM.WW.MxGateway.Worker.Bootstrap.IWorkerLogger { private readonly object gate = new(); @@ -2700,3 +2711,19 @@ public sealed class WorkerPipeSessionTests } } } + +/// +/// Collection for tests that observe process-global state and so cannot share the runner with +/// anything else. Its only member today is +/// , which arms +/// and forces a GC: a task faulting in any +/// concurrently running test class would be finalized inside that window and misread as this +/// session's orphaned pipe read. Keep membership minimal — every test added here is a test the +/// rest of the suite has to wait for. +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class WorkerPipeSessionNonParallelCollection +{ + /// Collection name referenced by . + public const string Name = "WorkerPipeSessionNonParallel"; +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index 7f1eb3e..f849813 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -167,18 +167,24 @@ public sealed class WorkerPipeSession // Closing the transport is what actually ends a pipe read parked in the kernel: on net48 // NamedPipeClientStream.ReadAsync ignores its CancellationToken, so the message loop's - // cancellation can never reach one (WRK-31). This is deliberately the LAST teardown step, - // because every frame this session will ever write has completed by the time control - // reaches here: WorkerFrameWriter.WriteAsync signals only after the frame is written AND - // flushed, and every exit path awaits its final write before unwinding — the shutdown ack - // and shutdown-timeout fault inside the loop's dispatch, the event-drain and - // oversized-event faults inside the drain task the loop awaits, the watchdog fault inside - // the heartbeat task the loop awaits, and the handshake fault inside - // CompleteStartupHandshakeAsync's catch. In-flight command replies are the one class of - // write that can still be racing here, and they raced the identical disposal before this - // change (WorkerPipeClient's `using` fired on the very next statement after RunAsync); - // both ProcessCommandAsync's Ready-state gate and TryWriteFaultAsync's - // ObjectDisposedException/IOException swallow already cover that race. + // cancellation can never reach one (WRK-31). It is deliberately the LAST teardown step, + // because in the ordinary case every frame this session will ever write has completed by + // the time control reaches here: WorkerFrameWriter.WriteAsync signals only after the + // frame is written AND flushed, and each exit path awaits its final write before + // unwinding — the shutdown ack and shutdown-timeout fault inside the loop's dispatch, the + // event-drain and oversized-event faults inside the drain task the loop awaits, the + // watchdog fault inside the heartbeat task the loop awaits, and the handshake fault + // inside CompleteStartupHandshakeAsync's catch. + // + // "Ordinary" is the honest word, not "always": the loop's wait on the heartbeat and + // drain tasks is budgeted (BackgroundTaskStopTimeout), and a stream write is genuinely + // uncancellable, so a write that overran the budget can still be in flight against the + // stream being disposed here. In-flight command replies are in the same position, and + // they raced the identical disposal before this change (WorkerPipeClient's `using` fired + // on the very next statement after RunAsync). That is precisely why disposal below is + // exception-tolerant and why every abandoned task gets a fault-observing continuation + // from ObserveBackgroundTaskStopAsync — a write losing its stream mid-flight must be a + // logged non-event, not a lost terminal exception or an unobserved task. // // Owning the disposal here — rather than leaving it to WorkerPipeClient's `using` — is // what makes the abandoned read observable: the fault it takes on disposal lands on a @@ -191,12 +197,21 @@ public sealed class WorkerPipeSession /// /// Disposes the transport this session owns, if it was handed one. Dispose-time failures are - /// logged and swallowed: this runs inside 's finally, where letting an - /// escape would replace the exception that actually ended the - /// session (a shutdown timeout, a protocol violation, an event too large to frame) with a - /// far less actionable one. Disposal is idempotent, so WorkerPipeClient's outer - /// using re-disposing the same stream immediately afterwards is a no-op. + /// logged and swallowed: this runs inside 's finally, where letting a + /// failure escape would replace the exception that actually ended the session (a shutdown + /// timeout, a protocol violation, an event too large to frame) with a far less actionable + /// one. Disposal is idempotent, so WorkerPipeClient's outer using re-disposing + /// the same stream immediately afterwards is a no-op. /// + /// + /// The catch is deliberately total rather than the / + /// pair the fault-write paths use, matching + /// 's shape. Narrowing it to the expected types + /// would let an unexpected one — a Win32Exception surfaced by the handle close, say — + /// do the exact harm this guard exists to prevent. The rule is about the position in the + /// code, not about which exceptions are plausible: nothing thrown while releasing a handle + /// is more actionable than the session's terminal exception, so nothing may displace it. + /// private void DisposeTransportStream() { if (_transportStream is null) @@ -208,7 +223,7 @@ public sealed class WorkerPipeSession { _transportStream.Dispose(); } - catch (Exception exception) when (exception is IOException || exception is ObjectDisposedException) + catch (Exception exception) { _logger?.Error( "WorkerPipeSessionTransportDisposeFailed", @@ -226,16 +241,26 @@ public sealed class WorkerPipeSession /// event-drain loops, so every fault exit (event-drain fault, oversized event, heartbeat /// write failure) leaves a read outstanding on a task the loop never awaits again. Once /// has closed the handle that read faults with - /// or ; awaiting it here is - /// what keeps the fault observed, because the worker installs no + /// , , or a zero-byte read + /// mapped to EndOfStream, and that fault has to be observed — the worker installs no /// TaskScheduler.UnobservedTaskException handler. /// /// - /// No second read can follow this one. The message loop only ever issues a read after - /// awaiting the previous one, and it never re-enters after unwinding, so - /// 's single-consumer invariant — and with it the safety of - /// its reused length-prefix buffer and its pooled payload buffer, which the abandoned read - /// still owns until it faults — holds through teardown. + /// + /// The fault is observed unconditionally; it is only logged within + /// . Windows is under no obligation to deliver + /// the abandoned read's completion inside that budget, so a bounded await alone would + /// reopen the orphaning window it was added to close. + /// therefore hands the task a fault-observing continuation when it gives up waiting, + /// which makes the budget a diagnostics decision rather than a correctness one. + /// + /// + /// No second read can follow this one. The message loop only ever issues a read after + /// awaiting the previous one, and it never re-enters after unwinding, so + /// 's single-consumer invariant — and with it the safety + /// of its reused length-prefix buffer and its pooled payload buffer, which the abandoned + /// read still owns until it faults — holds through teardown. + /// /// /// A task that represents the asynchronous operation. private async Task ObserveAbandonedPipeReadAsync() @@ -405,6 +430,9 @@ public sealed class WorkerPipeSession .CreateLinkedTokenSource(cancellationToken); Task heartbeatTask = RunHeartbeatLoopAsync(heartbeatCancellation.Token); Task eventDrainTask = RunEventDrainLoopAsync(heartbeatCancellation.Token); + Debug.Assert( + _pendingReadTask is null, + "A frame read must never be issued while another is outstanding: WorkerFrameReader is single-consumer, and a second read would race the abandoned one for the reused prefix buffer and could return a pooled payload buffer twice."); Task readTask = _reader.ReadAsync(loopCancellation.Token); _pendingReadTask = readTask; @@ -425,6 +453,9 @@ public sealed class WorkerPipeSession return; } + Debug.Assert( + _pendingReadTask is null, + "The previous read must have been awaited before the next is issued: WorkerFrameReader is single-consumer."); readTask = _reader.ReadAsync(loopCancellation.Token); _pendingReadTask = readTask; } @@ -447,6 +478,32 @@ public sealed class WorkerPipeSession } } + /// + /// Waits a bounded time for a background task to stop and records what happened. + /// + /// + /// + /// The budget bounds the logging, never the observation. Every task this is + /// asked to observe is uncancellable at the point that matters — a net48 pipe read + /// ignores its token outright, and WorkerFrameWriter.WriteFrameAsync issues the + /// stream write under CancellationToken.None so a frame is never left + /// half-written on the wire — so any of them can outlive the budget and only then fault, + /// typically against a transport has since disposed. Overrunning + /// the budget and returning is therefore not enough: the task would be left with nobody + /// holding it, which is the exact orphaning this method exists to prevent. + /// + /// + /// So the timeout path hands the task a fault-observing continuation before returning. + /// The fault is then observed unconditionally, whenever it arrives; the budget only + /// decides whether it also gets logged here or is swallowed silently by the + /// continuation. That distinction matters because the worker installs no + /// TaskScheduler.UnobservedTaskException handler, so an unheld faulted task + /// surfaces only at finalization. + /// + /// + /// Background task being stopped. + /// Name recorded in the diagnostic logs. + /// A task that represents the asynchronous operation. private async Task ObserveBackgroundTaskStopAsync( Task task, string taskName) @@ -456,6 +513,7 @@ public sealed class WorkerPipeSession .ConfigureAwait(false); if (completedTask != task) { + ObserveFaultWhenever(task); _logger?.Error( "WorkerPipeSessionBackgroundTaskStopTimedOut", new Dictionary @@ -485,6 +543,24 @@ public sealed class WorkerPipeSession } } + /// + /// Attaches a continuation that observes 's exception whenever the + /// task eventually faults, so a task nobody is awaiting any more can never reach the + /// finalizer with an unobserved exception. Mirrors the shape + /// WorkerFrameWriter.ObserveAbandonedFault uses for frames a cancelled caller stops + /// awaiting (NEXT-04). Faulting is the only outcome that runs the continuation, and the + /// continuation is scheduled inline, so this costs nothing on the ordinary path. + /// + /// Task that may fault after its awaiter has walked away. + private static void ObserveFaultWhenever(Task task) + { + _ = task.ContinueWith( + static faultedTask => _ = faultedTask.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + private async Task RunEventDrainLoopAsync(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested)