dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 14 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix(worker): unconditional fault observation for abandoned pipe I/O; exception-total transport dispose

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 21:21:25 -04:00
parent e913dab5db
commit 3ef56be2dd
3 changed files with 262 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs
+133 -106
View File
@@ -912,112 +912,6 @@ public sealed class WorkerPipeSessionTests
await Assert.ThrowsAsync<InvalidOperationException>(async () => await runTask);
}
/// <summary>
/// WRK-31. A fault exit unwinds the message loop while its frame read is still pending, and
/// on net48 nothing can cancel that read — <c>NamedPipeClientStream.ReadAsync</c> 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
/// <c>TaskScheduler.UnobservedTaskException</c> 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task RunAsync_WhenFaultEndsSession_ObservesTheAbandonedPipeRead()
{
const uint tinyMaxFrameBytes = 4096;
object unobservedGate = new();
List<Exception> unobservedPipeExceptions = new();
EventHandler<UnobservedTaskExceptionEventArgs> 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<InvalidOperationException>(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);
}
}
/// <summary>
/// WRK-31, the other side of the invariant. The graceful path leaves the message loop
/// through its <c>return</c> after the shutdown ack, with that iteration's read already
@@ -2381,6 +2275,123 @@ public sealed class WorkerPipeSessionTests
return envelopes.ToArray();
}
/// <summary>
/// The one teardown test that has to arm <see cref="TaskScheduler.UnobservedTaskException"/>,
/// 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 <see cref="WorkerPipeSessionNonParallelCollection"/>) rather
/// than alongside its siblings. Nested so it can still reach
/// <see cref="WorkerPipeSessionTests"/>'s private harness — <c>PipePair</c>,
/// <c>CreatePipeSession</c>, <c>RecordingWorkerLogger</c> — without widening any of it.
/// </summary>
[Collection(WorkerPipeSessionNonParallelCollection.Name)]
public sealed class AbandonedPipeReadTeardownTests
{
/// <summary>
/// WRK-31. A fault exit unwinds the message loop while its frame read is still pending,
/// and on net48 nothing can cancel that read — <c>NamedPipeClientStream.ReadAsync</c>
/// 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 <c>TaskScheduler.UnobservedTaskException</c> 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task RunAsync_WhenFaultEndsSession_ObservesTheAbandonedPipeRead()
{
const uint tinyMaxFrameBytes = 4096;
object unobservedGate = new();
List<Exception> unobservedPipeExceptions = new();
EventHandler<UnobservedTaskExceptionEventArgs> 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<InvalidOperationException>(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
}
}
}
/// <summary>
/// Collection for tests that observe process-global state and so cannot share the runner with
/// anything else. Its only member today is
/// <see cref="WorkerPipeSessionTests.AbandonedPipeReadTeardownTests"/>, which arms
/// <see cref="TaskScheduler.UnobservedTaskException"/> 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.
/// </summary>
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class WorkerPipeSessionNonParallelCollection
{
/// <summary>Collection name referenced by <see cref="CollectionAttribute"/>.</summary>
public const string Name = "WorkerPipeSessionNonParallel";
}