diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index e2f25b9..ebff8d7 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -171,13 +171,33 @@ oversized event) surfaces from the batch's awaited completions as that frame's `WorkerFrameProtocolException`; the remaining completions are still observed so none faults unobserved. -The completion is the frame's delivery point, not necessarily the instant its -caller returns. A caller that loses the race for the write lock only observes -its own completion after the winning drainer releases the lock, so its return -remains bounded by that drain pass even though its control frame was flushed -and completed at the class boundary inside it. The boundary flush is what -makes the delivery point honest; unparking a lock-race loser from the winner's -pass would be a separate change to the enqueue-then-contend shape. +The completion is the frame's delivery point, and a `WriteAsync` caller now +returns at it. The boundary flush alone only made the delivery point honest: +a caller that lost the race for the write lock still sat in the lock wait +until the winning drainer released it, so its awaited task was charged for the +whole event backlog its control frame had just been flushed ahead of. To close +that, `WriteAsync` awaits its own frame's completion *racing* the lock +acquisition instead of the acquisition alone. Whichever settles first decides: + +- **Completion first** — the winning drainer wrote and flushed this frame at + the class boundary, so the caller returns immediately. The lock acquisition + it leaves outstanding is *detached*, not dropped: a continuation drains + whatever is queued and then releases, so the lock is never acquired and + silently held, and a frame enqueued between the previous drainer's last + dequeue and its release is still written by someone. Draining an empty queue + is a no-op, so the common case is acquire-nothing-release. +- **Lock first** — the caller drains the pass itself, exactly as before. +- **Cancellation** — the wait ends without the lock (`SemaphoreSlim` hands no + count to a wait it cancels, so the detached continuation releases nothing on + that path) and the tombstone rules below apply unchanged. A token that fires + *after* the frame's completion won the race changes nothing: the frame was + delivered, and the caller returns normally. + +`WriteBatchAsync` deliberately keeps the plain wait-then-drain shape. A batch +caller's result is its whole set of completions and the last of those resolves +at the end-of-pass flush — the instant before the drainer releases the lock — +so racing the acquisition would buy it nothing while adding one detached +acquisition per call. Cancellation of a `WriteAsync`/`WriteBatchAsync` call that is still waiting for the write lock when its token fires tombstones the queued frame: the diff --git a/docs/plans/2026-08-15-deferred-remediation.md b/docs/plans/2026-08-15-deferred-remediation.md index a22e610..4da7f6c 100644 --- a/docs/plans/2026-08-15-deferred-remediation.md +++ b/docs/plans/2026-08-15-deferred-remediation.md @@ -349,6 +349,9 @@ boundary, so the priority class governs the frame's delivery point rather than o its byte order. Getting the awaited-latency win too requires unparking the lock-race loser from the winner's pass — a change to the write-lock shape, recorded as a follow-up. One extra `FlushFileBuffers` per mixed pass is the accepted cost. +That lock-parking was closed by `docs/plans/2026-08-17-deferred-closeout.md` +Task 2, 2026-08-17: `WriteAsync` races its own frame's completion against the +lock acquisition and detaches the wait it abandons. **Task 11 — teardown ordering and unconditional fault observation.** Teardown disposes the session-owned transport first, then observes the read that dispose abandoned. 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 a58fed6..9426325 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -428,10 +428,10 @@ public sealed class WorkerFrameProtocolTests /// control run has already run, so the heartbeat or reply is on the pipe rather than waiting behind /// the batch. Exactly two flushes for the pass — one per class run, not one per control frame. /// - /// The assertion is on the flush, not on the queued callers' returned tasks, because those tasks are - /// still gated by the write lock they lost to the drainer (see the latency contract on - /// WorkerFrameWriter.WriteAsync): the completion resolves at the boundary flush, but a - /// lock-race loser observes it only once the drainer releases the lock. + /// The assertion here is on the flush and on the event callers, whose delivery point is + /// still the end-of-pass flush. That the control frame's own caller returns at the boundary — the + /// lock-parking this pass used to impose on it — is the separate subject of + /// . /// /// /// A task that represents the asynchronous operation. @@ -912,6 +912,201 @@ public sealed class WorkerFrameProtocolTests Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f4.BodyCase); } + /// + /// Frame-writer lock-parking, closed. The class-boundary flush made a control frame's delivery + /// point honest, but the caller that lost the write-lock race still could not observe it: it + /// sat in WaitAsync until the winning drainer released the lock, so its awaited task was + /// charged for the whole event backlog the boundary flush had just jumped it ahead of. The caller + /// now races its own frame's completion against the lock acquisition, so it returns at the boundary. + /// + /// The winner here is a WriteBatchAsync event burst — the production hot path, the event + /// drain loop's own call — gated so the pass is stopped inside the batch, after the boundary flush + /// that delivered the control frame and long before the pass ends. The control caller returning + /// while the drainer is demonstrably still blocked mid-batch is the whole property; under the parked + /// shape this await could not return until ReleaseSecondGateWrite, so a regression shows up + /// as AwaitWithTimeoutAsync's rather than as a hang. + /// + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_ControlFrameLosingLockRaceToEventBatch_ReturnsAtItsOwnCompletion() + { + WorkerFrameProtocolOptions options = CreateOptions(); + // Write 1 (the batch's first event) gates the pass open; write 3 is the first event written + // after the control frame's boundary flush, so blocking it stops the drain with the control + // frame delivered and the rest of the batch still unwritten. + using GatedWriteStream stream = new(secondGateWriteIndex: 3); + WorkerFrameWriter writer = new(stream, options); + + WorkerEnvelope[] batch = new[] + { + CreateEventEnvelope(workerSequence: 1), + CreateEventEnvelope(workerSequence: 2), + CreateEventEnvelope(workerSequence: 3), + CreateEventEnvelope(workerSequence: 4), + }; + + Task batchWrite = writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + // The control frame loses the lock race to the batch and is drained by it. + Task controlWrite = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted); + + // The boundary flush ran and the drain is now blocked inside the batch behind it. + Assert.Equal(1, stream.FlushCount); + Assert.False(batchWrite.IsCompleted); + + // The latency win: the loser returns here, with the winner's pass still in flight. + await AwaitWithTimeoutAsync(controlWrite); + Assert.False(batchWrite.IsCompleted); + + stream.ReleaseSecondGateWrite(); + await AwaitWithTimeoutAsync(batchWrite); + + // One flush per class run, unchanged: the control run's boundary flush, then the batch's. + Assert.Equal(2, stream.FlushCount); + + // Detaching the abandoned lock wait strands nothing: every frame is on the wire exactly once, + // in drain order, with contiguous write-time sequences. + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + WorkerEnvelope frame3 = await reader.ReadAsync(); + WorkerEnvelope frame4 = await reader.ReadAsync(); + WorkerEnvelope frame5 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame4.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame5.BodyCase); + Assert.Equal( + new ulong[] { 1, 2, 3, 4, 5 }, + new[] { frame1.Sequence, frame2.Sequence, frame3.Sequence, frame4.Sequence, frame5.Sequence }); + Assert.Equal(stream.Length, stream.Position); + } + + /// + /// Frame-writer lock-parking, the "nothing is stranded" half. A caller that returns on its + /// completion abandons a live write-lock acquisition; that acquisition still carries drain + /// responsibility, because a frame can be enqueued after the winning drainer's last dequeue and + /// before its release. Under heavy mixed-priority concurrency — every caller racing its completion + /// against the lock, so detached acquisitions pile up — every frame must still be written exactly + /// once, and the write lock must still admit exactly one drainer: a double-drain would interleave + /// two passes over the same stream, and a double-release would either do that or throw + /// out of a later drain. Contiguous 1..N sequences with no + /// duplicates and no trailing bytes is the observable form of both. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_UnderMixedPriorityConcurrency_WritesEveryQueuedFrameExactlyOnce() + { + const int perClass = 40; + WorkerFrameProtocolOptions options = CreateOptions(); + using MemoryStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + Task[] writes = new Task[perClass * 2]; + for (int index = 0; index < perClass; index++) + { + writes[index * 2] = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + writes[(index * 2) + 1] = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + } + + await AwaitWithTimeoutAsync(Task.WhenAll(writes)); + + // A detached acquisition drains whatever it finds and releases; this write goes through the + // same lock afterwards, so it can only succeed if the lock was left in a usable state. + await AwaitWithTimeoutAsync( + writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control)); + + const int total = (perClass * 2) + 1; + int controlCount = 0; + int eventCount = 0; + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + for (int index = 0; index < total; index++) + { + WorkerEnvelope frame = await reader.ReadAsync(); + Assert.Equal((ulong)(index + 1), frame.Sequence); + if (frame.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerEvent) + { + eventCount++; + } + else + { + controlCount++; + } + } + + Assert.Equal(perClass + 1, controlCount); + Assert.Equal(perClass, eventCount); + // No frame was written twice and none was left queued. + Assert.Equal(stream.Length, stream.Position); + } + + /// + /// Frame-writer lock-parking, the cancellation corner. A caller that returned on its completion + /// leaves a live lock acquisition behind; if its token then fires, that acquisition is cancelled + /// after the caller is long gone. hands no count to a wait it cancels, + /// so the detached continuation must release nothing on that path — releasing there would push the + /// count past the maximum and make the drainer's own Release throw + /// , which is exactly what awaiting the drainer's write here + /// detects. The late cancellation must also not retroactively cancel the call that already + /// returned, nor tombstone a frame that is already on the wire. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_TokenCancelledAfterCompletionFirstReturn_LeavesTheWriteLockIntact() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(secondGateWriteIndex: 3); + WorkerFrameWriter writer = new(stream, options); + + // The drainer holds the lock, blocked writing its own control frame. + Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + using CancellationTokenSource cts = new(); + Task lateControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control, cts.Token); + Task eventWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + await Task.Delay(50); + + // The drain writes both control frames, flushes them at the class boundary, then blocks on the + // event write — so the cancellable caller returns on its completion with its acquisition live. + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted); + await AwaitWithTimeoutAsync(lateControl); + + // Cancel the acquisition nobody is waiting on any more. + cts.Cancel(); + + stream.ReleaseSecondGateWrite(); + + // A count released on the cancelled path would surface here, as the drainer's release throwing. + await AwaitWithTimeoutAsync(Task.WhenAll(firstControl, eventWrite)); + + // The lock is still usable, and the cancelled token did not recall the delivered frame. + await AwaitWithTimeoutAsync( + writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control)); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + WorkerEnvelope frame3 = await reader.ReadAsync(); + WorkerEnvelope frame4 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame4.BodyCase); + Assert.Equal(stream.Length, stream.Position); + } + private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1) { return new WorkerEnvelope diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index d94b118..c8a1e9a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -16,9 +16,12 @@ namespace ZB.MOM.WW.MxGateway.Worker.Ipc; /// holds the lock drains every queued frame, control frames first, so a reply, fault, or heartbeat is /// never delayed behind an event backlog — neither in the bytes it writes nor in the flush that /// delivers them, because the drain flushes at every control-to-event boundary rather than only at the -/// end of the pass. The envelope Sequence is stamped by the draining lock-holder at the moment -/// of writing, so the on-wire order and the stamped sequence always agree even under concurrent callers -/// and priority reordering. +/// end of the pass. A caller that loses the lock race does not wait for the winner's pass to end: it +/// awaits its own frame's completion racing its lock acquisition, so it returns at the boundary flush +/// that delivered its frame and the acquisition it walks away from is detached rather than dropped +/// (see ). The envelope Sequence is stamped by the draining +/// lock-holder at the moment of writing, so the on-wire order and the stamped sequence always agree +/// even under concurrent callers and priority reordering. /// public sealed class WorkerFrameWriter { @@ -111,10 +114,12 @@ public sealed class WorkerFrameWriter /// Latency contract: a control frame's bytes are written, flushed, and its completion /// resolved before the events a drain pass writes after it — the delivery point of a heartbeat, /// reply, fault, or shutdown ack is never charged for the event backlog behind it. The returned - /// task can still be later than that instant for a caller that lost the write-lock race: it only - /// observes its completion after the winning drainer releases the lock, so its own return remains - /// bounded by that pass. That parking is deliberate — the alternative is to race the lock wait - /// against the completion, which buys nothing for the frame's delivery. + /// task resolves at that same instant even for a caller that lost the write-lock race, because + /// this call awaits its own frame's completion racing the lock acquisition rather than the lock + /// alone: the completion resolving first returns the caller immediately and hands the outstanding + /// acquisition to , which keeps its drain responsibility. Without that, + /// a control frame delivered at a class boundary still reported back only after the winning + /// drainer finished writing and flushing the entire event backlog behind it. /// /// public async Task WriteAsync( @@ -140,20 +145,39 @@ public sealed class WorkerFrameWriter } } - // Contend for the single writer: whoever wins drains every currently-queued frame in priority - // order, so this frame is written by this call or by a concurrent caller that got the lock - // first. Either way it completes via its own TaskCompletionSource. - try + // Contend for the single writer, but race that contention against this frame's own completion. + // Whoever wins the lock drains every currently-queued frame in priority order, so this frame is + // written by this call or by a concurrent caller that got the lock first — and in the latter + // case the winner writes, flushes, and completes it at the control-to-event boundary, part-way + // through its pass. Racing the two is what lets this call return at that instant instead of at + // the winner's release; the acquisition it then walks away from is detached, never dropped, so + // no drain responsibility leaves with it. + Task lockWait = _writeLock.WaitAsync(cancellationToken); + Task completion = frame.Completion.Task; + await Task.WhenAny(completion, lockWait).ConfigureAwait(false); + + if (lockWait.Status != TaskStatus.RanToCompletion) { - await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); - } - catch (OperationCanceledException) - { - // Tombstone the queued frame so DequeueNext skips it — but only if a draining lock-holder - // has not already claimed it. If it is claimed it is mid-write and cannot be recalled; the - // caller still observes cancellation while the frame reaches the wire (documented above). - TombstoneIfUnclaimed(frame, cancellationToken); - throw; + // This call does not hold the lock: either the completion won the race and the acquisition + // is still outstanding, or the wait ended without the lock because the token fired. Hand the + // wait over either way — an acquisition nobody is waiting on must still drain and release, + // and a wait that ends without the lock must still have its outcome observed. + DetachLockWait(lockWait); + + if (!completion.IsCompleted) + { + // The frame was not delivered, so the wait must have ended in cancellation. Tombstone + // the queued frame so DequeueNext skips it — but only if a draining lock-holder has not + // already claimed it. If it is claimed it is mid-write and cannot be recalled; the + // caller still observes cancellation while the frame reaches the wire (documented + // above). Rethrow from the wait rather than from the completion, so a claimed frame's + // canceller is not held behind the very write it is abandoning. + TombstoneIfUnclaimed(frame, cancellationToken); + await lockWait.ConfigureAwait(false); + } + + await completion.ConfigureAwait(false); + return; } try @@ -165,7 +189,7 @@ public sealed class WorkerFrameWriter _writeLock.Release(); } - await frame.Completion.Task.ConfigureAwait(false); + await completion.ConfigureAwait(false); } /// @@ -195,6 +219,13 @@ public sealed class WorkerFrameWriter /// ; 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). + /// + /// This path deliberately keeps the plain wait-then-drain shape rather than the single-frame + /// path's completion-versus-lock race. A batch caller's result is its whole set of completions, + /// and the last of those resolves at the end-of-pass flush — the instant before the drainer + /// releases the lock — so racing the acquisition would buy a batch caller nothing while adding a + /// detached acquisition per call. Semantics here are unchanged by that race. + /// /// public async Task WriteBatchAsync( IReadOnlyList envelopes, @@ -319,6 +350,79 @@ public sealed class WorkerFrameWriter TaskScheduler.Default); } + /// + /// Keeps a write-lock acquisition whose caller has stopped waiting for it — its frame was + /// delivered inside the winning drainer's pass, or its token fired — from losing the drain + /// responsibility that comes with the lock. The wait is never simply dropped: if it goes on to + /// acquire the lock, the continuation drains whatever is queued and then releases, so the lock + /// is never acquired and silently held, and a frame enqueued between the previous drainer's + /// last and its release is still written by someone. A wait that ends + /// without the lock took no semaphore count and so has nothing to release. + /// + /// The continuation is queued to the thread pool rather than run inline, because it starts a + /// drain pass: running that synchronously would charge whichever thread called Release + /// for the next caller's writes. + /// + /// + /// Outstanding write-lock acquisition the caller has walked away from. + private void DetachLockWait(Task lockWait) + { + _ = lockWait.ContinueWith( + OnDetachedLockWaitSettled, + CancellationToken.None, + TaskContinuationOptions.None, + TaskScheduler.Default); + } + + /// + /// Settles a detached write-lock acquisition: drain and release if it took the lock, otherwise + /// observe its outcome and stop. Acquisition and cancellation are mutually exclusive — + /// never hands a count to a wait it cancels — so neither branch can + /// leak a count, and neither can strand a queued frame: a cancelled wait never held the lock, + /// so whatever is queued is still owned by the next caller to acquire it. + /// + /// Settled write-lock acquisition task. + private void OnDetachedLockWaitSettled(Task lockWait) + { + if (lockWait.Status != TaskStatus.RanToCompletion) + { + // Cancelled by the originating caller's token, or — only pathologically, a disposed + // semaphore — faulted. No count was taken, so there is nothing to release and no drain to + // inherit. Touch Exception so a fault on a task nobody awaits any more cannot surface as an + // unobserved-task exception. + _ = lockWait.Exception; + return; + } + + _ = DrainDetachedAsync(); + } + + /// + /// Runs a drain pass under a write lock this writer acquired on behalf of a caller that has + /// already returned, then releases it. An empty queue makes + /// a no-op, so the common case is acquire-drain-nothing-release; the pass earns its keep for a + /// frame enqueued after the previous drainer's last dequeue but before its release. + /// + /// A task that completes once the drain pass has ended and the lock has been released. + private async Task DrainDetachedAsync() + { + try + { + await DrainQueuedFramesAsync().ConfigureAwait(false); + } + catch (Exception) + { + // DrainQueuedFramesAsync routes every write and flush failure onto the affected frames' + // completions, so nothing is expected to escape it. If anything ever does, there is no + // caller left on this pass to receive it and letting it out would only raise an + // unobserved-task exception. The release below is the part that must not be skipped. + } + finally + { + _writeLock.Release(); + } + } + // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each. // The stream write itself is not cancellable: a frame is written atomically or fails, never left // half-written on the pipe because a caller gave up waiting.