From 84dbf20a43288c311b65c35bc0f44a1d4f0d4f5d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 10 Aug 2026 05:57:53 -0400 Subject: [PATCH] fix(worker): observe faults on frames abandoned by cancellation (NEXT-04, NEXT-05 decision) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A WriteAsync/WriteBatchAsync caller cancelled after the draining lock-holder claimed its frame unwinds without awaiting that frame's completion; the same holds for a frame already faulted by a concurrent FailAllQueued, where TrySetCanceled loses. A later wire-write failure then lands TrySetException on a task with no awaiter and surfaces as TaskScheduler.UnobservedTaskException. The tombstone helpers now attach a fault-observing continuation to every frame in the cancelled call (a cancelled task never fires OnlyOnFaulted, so unconditional attach is safe), outside _gate because an already-faulted task runs the continuation inline. NEXT-05 is resolved as a documented decision, not a code change: tombstoned entries keep their lazy DequeueNext purge — any subsequent write drains both queues to empty and the heartbeat loop bounds residency to one interval, while eager Queue rebuilds under _gate would add ordering-invariant surface for no gain. Rationale recorded in docs/WorkerFrameProtocol.md alongside the WRK-22 residual-window contract. New regression test drives the exact abandonment: gated stream holds writer A mid-write, the queued event frame is claimed and blocked mid-write, its caller is cancelled, the write then faults with a marker exception, and the test asserts the marker never reaches UnobservedTaskException after a forced GC. net48 x86 build/test runs on windev with the rest of this batch. --- docs/WorkerFrameProtocol.md | 13 ++ .../Ipc/WorkerFrameProtocolTests.cs | 127 ++++++++++++++++++ .../Ipc/WorkerFrameWriter.cs | 32 ++++- 3 files changed, 171 insertions(+), 1 deletion(-) diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index a23631b..ea1aa59 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -147,6 +147,19 @@ so the caller observes `OperationCanceledException` while that one frame still reaches the wire. That residual window is by design: blocking the canceller behind the very write it is abandoning would defeat the point of cancellation. +Two hygiene notes on that residual (NEXT-04/NEXT-05). First, a frame the +cancelled caller abandons — claimed mid-write, or already faulted by a +concurrent queue-wide failure — completes on a task nobody awaits; the +tombstone path attaches a fault-observing continuation to it so a later write +failure never surfaces as a `TaskScheduler.UnobservedTaskException`. Second, +tombstoned entries stay in the class queues until a future `DequeueNext` pops +and skips them; that lazy purge is deliberate. Eagerly rebuilding a `Queue` +under `_gate` on every cancellation would add ordering-invariant surface next +to the claim/cancel interlock for no real gain: any subsequent write of either +class drains both queues to empty, and the heartbeat loop guarantees one +arrives within a heartbeat interval, so worst-case residency is a few envelope +references for seconds — not a leak. + ## Verification The frame protocol lives in `ZB.MOM.WW.MxGateway.Worker.Ipc` (`WorkerFrameReader`, diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs index 334e4a9..bdeea32 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -545,6 +545,70 @@ public sealed class WorkerFrameProtocolTests Assert.Equal(2UL, frame2.Sequence); } + /// + /// NEXT-04. A frame claimed by the draining lock-holder before its caller's cancellation lands + /// is abandoned — the cancelled caller never awaits its completion. If the wire write then + /// faults, the tombstone path's fault-observing continuation must still observe the exception + /// so it never surfaces as . + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_ClaimedFrameAbandonedByCancellation_FaultIsObserved() + { + string marker = $"NEXT-04-{Guid.NewGuid():N}"; + WorkerFrameProtocolOptions options = CreateOptions(); + bool sawUnobservedMarkerFault = false; + EventHandler handler = (sender, args) => + { + if (args.Exception.ToString().Contains(marker)) + { + sawUnobservedMarkerFault = true; + } + }; + + TaskScheduler.UnobservedTaskException += handler; + try + { + using (SecondWriteFaultingGatedStream stream = new SecondWriteFaultingGatedStream(marker)) + { + WorkerFrameWriter writer = new WorkerFrameWriter(stream, options); + + // Writer A holds the lock, blocked mid-write of its own frame. + Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + using (CancellationTokenSource cts = new CancellationTokenSource()) + { + // Queue the doomed event write behind A, release A so its drain claims the + // event frame and blocks mid-write of it, then cancel the queued caller — + // the frame is claimed, so the caller unwinds without an awaiter for it. + Task abandonedWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token); + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(stream.SecondWriteStarted); + cts.Cancel(); + await Assert.ThrowsAnyAsync(async () => await abandonedWrite); + } + + // Fault the abandoned frame's wire write; observe writer A's own outcome so only + // the abandoned frame's completion could ever raise the marker unobserved. + stream.ReleaseSecondWrite(); + _ = await Record.ExceptionAsync(async () => await firstWrite); + } + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + } + finally + { + TaskScheduler.UnobservedTaskException -= handler; + } + + Assert.False( + sawUnobservedMarkerFault, + "The abandoned frame's write fault surfaced as an unobserved-task exception."); + } + /// /// WRK-22 / IPC-26, the review's shutdown scenario. A cancelled event frame queued before a /// shutdown-ack control frame must not trail the ack on the wire: the tombstone rule plus the @@ -749,6 +813,69 @@ public sealed class WorkerFrameProtocolTests } } + // A MemoryStream whose first write blocks until released and whose second write blocks until + // released and then throws, so a test can abandon a claimed frame by cancellation and fault its + // wire write afterwards (NEXT-04). + private sealed class SecondWriteFaultingGatedStream : MemoryStream + { + private readonly SemaphoreSlim _firstRelease = new SemaphoreSlim(0); + private readonly SemaphoreSlim _secondRelease = new SemaphoreSlim(0); + private readonly TaskCompletionSource _firstWriteStarted = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _secondWriteStarted = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly string _faultMessage; + private int _writeCount; + + public SecondWriteFaultingGatedStream(string faultMessage) + { + _faultMessage = faultMessage; + } + + /// Gets a task that completes once the first call has started blocking. + public Task FirstWriteStarted => _firstWriteStarted.Task; + + /// Gets a task that completes once the second call has started blocking. + public Task SecondWriteStarted => _secondWriteStarted.Task; + + /// Releases the first blocked write so it can complete. + public void ReleaseFirstWrite() => _firstRelease.Release(); + + /// Releases the second blocked write so it can throw. + public void ReleaseSecondWrite() => _secondRelease.Release(); + + /// + public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + int writeIndex = Interlocked.Increment(ref _writeCount); + if (writeIndex == 1) + { + _firstWriteStarted.TrySetResult(true); + await _firstRelease.WaitAsync(cancellationToken); + } + else if (writeIndex == 2) + { + _secondWriteStarted.TrySetResult(true); + await _secondRelease.WaitAsync(cancellationToken); + throw new IOException(_faultMessage); + } + + await base.WriteAsync(buffer, offset, count, cancellationToken); + } + + /// + protected override void Dispose(bool disposing) + { + if (disposing) + { + _firstRelease.Dispose(); + _secondRelease.Dispose(); + } + + base.Dispose(disposing); + } + } + // A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames // behind an in-progress write and observe the writer's priority ordering. private sealed class GatedWriteStream : MemoryStream diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index 6061f00..7fab801 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -92,6 +92,8 @@ public sealed class WorkerFrameWriter /// already claimed it, in which case the frame may still reach the wire even though this call /// observes . That residual window is by design: blocking /// the canceller behind the very write it is abandoning would defeat the point of cancellation. + /// The abandoned frame's completion gets a fault-observing continuation so a write failure after + /// the caller unwinds never raises an unobserved-task exception (NEXT-04). /// public async Task WriteAsync( WorkerEnvelope envelope, @@ -162,7 +164,9 @@ public sealed class WorkerFrameWriter /// awaited completions as its ; the remaining frames are /// still observed so none faults unobserved. Cancellation while waiting for the lock tombstones /// every still-unclaimed frame in the batch, per the WRK-22 contract on - /// . + /// ; frames + /// the cancelled caller abandons (claimed mid-write, or already faulted) get a fault-observing + /// continuation so a later write failure never raises an unobserved-task exception (NEXT-04). /// public async Task WriteBatchAsync( IReadOnlyList envelopes, @@ -245,6 +249,8 @@ public sealed class WorkerFrameWriter frame.Completion.TrySetCanceled(cancellationToken); } } + + ObserveAbandonedFault(frame); } private void TombstoneUnclaimed(PendingFrame[] frames, CancellationToken cancellationToken) @@ -259,6 +265,30 @@ public sealed class WorkerFrameWriter } } } + + foreach (PendingFrame frame in frames) + { + ObserveAbandonedFault(frame); + } + } + + /// + /// Observes any fault on a frame the cancelled caller stops awaiting (NEXT-04). A frame + /// claimed by a draining lock-holder — or already faulted by a concurrent + /// FailAllQueued — completes on a task nobody awaits after cancellation unwinds the + /// caller; a later write failure would then surface as an unobserved-task exception. A + /// cancelled task never triggers the faulted continuation, so attaching unconditionally is + /// safe. Attached outside _gate because an already-faulted task runs the + /// continuation inline. + /// + /// Frame whose completion may fault without an awaiter. + private static void ObserveAbandonedFault(PendingFrame frame) + { + _ = frame.Completion.Task.ContinueWith( + task => _ = task.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); } // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each.