fix(worker): observe faults on frames abandoned by cancellation (NEXT-04, NEXT-05 decision)

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<T> 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.
This commit is contained in:
Joseph Doherty
2026-08-10 05:57:53 -04:00
parent 8769ee9765
commit 84dbf20a43
3 changed files with 171 additions and 1 deletions
+13
View File
@@ -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 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. 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<T>`
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 ## Verification
The frame protocol lives in `ZB.MOM.WW.MxGateway.Worker.Ipc` (`WorkerFrameReader`, The frame protocol lives in `ZB.MOM.WW.MxGateway.Worker.Ipc` (`WorkerFrameReader`,
@@ -545,6 +545,70 @@ public sealed class WorkerFrameProtocolTests
Assert.Equal(2UL, frame2.Sequence); Assert.Equal(2UL, frame2.Sequence);
} }
/// <summary>
/// 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 <see cref="TaskScheduler.UnobservedTaskException"/>.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WriteAsync_ClaimedFrameAbandonedByCancellation_FaultIsObserved()
{
string marker = $"NEXT-04-{Guid.NewGuid():N}";
WorkerFrameProtocolOptions options = CreateOptions();
bool sawUnobservedMarkerFault = false;
EventHandler<UnobservedTaskExceptionEventArgs> 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<OperationCanceledException>(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.");
}
/// <summary> /// <summary>
/// WRK-22 / IPC-26, the review's shutdown scenario. A cancelled event frame queued before a /// 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 /// 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<bool> _firstWriteStarted =
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly TaskCompletionSource<bool> _secondWriteStarted =
new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly string _faultMessage;
private int _writeCount;
public SecondWriteFaultingGatedStream(string faultMessage)
{
_faultMessage = faultMessage;
}
/// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
public Task FirstWriteStarted => _firstWriteStarted.Task;
/// <summary>Gets a task that completes once the second <see cref="WriteAsync"/> call has started blocking.</summary>
public Task SecondWriteStarted => _secondWriteStarted.Task;
/// <summary>Releases the first blocked write so it can complete.</summary>
public void ReleaseFirstWrite() => _firstRelease.Release();
/// <summary>Releases the second blocked write so it can throw.</summary>
public void ReleaseSecondWrite() => _secondRelease.Release();
/// <inheritdoc />
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);
}
/// <inheritdoc />
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 // 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. // behind an in-progress write and observe the writer's priority ordering.
private sealed class GatedWriteStream : MemoryStream private sealed class GatedWriteStream : MemoryStream
@@ -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 /// already claimed it, in which case the frame may still reach the wire even though this call
/// observes <see cref="OperationCanceledException"/>. That residual window is by design: blocking /// observes <see cref="OperationCanceledException"/>. That residual window is by design: blocking
/// the canceller behind the very write it is abandoning would defeat the point of cancellation. /// 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).
/// </remarks> /// </remarks>
public async Task WriteAsync( public async Task WriteAsync(
WorkerEnvelope envelope, WorkerEnvelope envelope,
@@ -162,7 +164,9 @@ public sealed class WorkerFrameWriter
/// awaited completions as its <see cref="WorkerFrameProtocolException"/>; the remaining frames are /// awaited completions as its <see cref="WorkerFrameProtocolException"/>; the remaining frames are
/// still observed so none faults unobserved. Cancellation while waiting for the lock tombstones /// 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 /// every still-unclaimed frame in the batch, per the WRK-22 contract on
/// <see cref="WriteAsync(WorkerEnvelope, WorkerFrameWritePriority, CancellationToken)"/>. /// <see cref="WriteAsync(WorkerEnvelope, WorkerFrameWritePriority, CancellationToken)"/>; 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).
/// </remarks> /// </remarks>
public async Task WriteBatchAsync( public async Task WriteBatchAsync(
IReadOnlyList<WorkerEnvelope> envelopes, IReadOnlyList<WorkerEnvelope> envelopes,
@@ -245,6 +249,8 @@ public sealed class WorkerFrameWriter
frame.Completion.TrySetCanceled(cancellationToken); frame.Completion.TrySetCanceled(cancellationToken);
} }
} }
ObserveAbandonedFault(frame);
} }
private void TombstoneUnclaimed(PendingFrame[] frames, CancellationToken cancellationToken) private void TombstoneUnclaimed(PendingFrame[] frames, CancellationToken cancellationToken)
@@ -259,6 +265,30 @@ public sealed class WorkerFrameWriter
} }
} }
} }
foreach (PendingFrame frame in frames)
{
ObserveAbandonedFault(frame);
}
}
/// <summary>
/// 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
/// <c>FailAllQueued</c> — 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 <c>_gate</c> because an already-faulted task runs the
/// continuation inline.
/// </summary>
/// <param name="frame">Frame whose completion may fault without an awaiter.</param>
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. // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each.