From e913dab5db69f4523f74a92a082e7fe9e9fb4ec1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 21:06:03 -0400 Subject: [PATCH] fix(worker): session-owned stream disposal unblocks and observes the net48 pipe read at teardown --- docs/MxAccessWorkerInstanceDesign.md | 30 +++ .../Ipc/WorkerPipeSessionTests.cs | 218 +++++++++++++++++- .../Ipc/WorkerFrameReader.cs | 14 ++ .../Ipc/WorkerPipeClient.cs | 7 + .../Ipc/WorkerPipeSession.cs | 142 +++++++++++- 5 files changed, 408 insertions(+), 3 deletions(-) diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index c194340..35c6cf1 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -887,6 +887,36 @@ Graceful shutdown sequence: If shutdown wedges, the gateway kills the process. The worker should be written so process kill does not corrupt other sessions. +### Ending the pipe read (net48) + +Step 8 above cannot be done by cancellation. On .NET Framework 4.8 +`NamedPipeClientStream.ReadAsync` accepts a `CancellationToken` and then never +wires it to the overlapped I/O, so a read parked waiting for gateway bytes stays +parked no matter what the worker cancels. Closing the handle is the only thing +that ends it. + +`WorkerPipeSession.RunMessageLoopAsync` races one outstanding read against the +heartbeat and event-drain loops, so every fault exit — an event-drain fault, an +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. + +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. + `MxAccessStaSession.ShutdownGracefullyAsync` implements the current cleanup path. It first calls `StaCommandDispatcher.RequestShutdown()` so new commands are rejected and queued commands that have not started receive 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 4f6703c..18580cd 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -912,6 +912,192 @@ 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 + /// awaited — so there is no abandoned read for teardown to account for. Pinning this keeps + /// the new disposal-and-observe step a pure no-op on the path production takes every time a + /// session closes normally: no "PipeRead" observation, and therefore no chance of paying + /// the observation timeout on a healthy shutdown. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_GracefulShutdown_LeavesNoPendingPipeReadToObserve() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + RecordingWorkerLogger logger = new(); + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(100), + HeartbeatGrace = TimeSpan.FromSeconds(5), + }, + logger); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + // Reads the ack and bounds RunAsync's completion at 5s. + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + + Assert.True(runtime.Disposed, "Graceful shutdown must dispose the runtime session."); + Assert.DoesNotContain( + logger.Events, + entry => (entry.EventName == "WorkerPipeSessionBackgroundTaskStopFailed" + || entry.EventName == "WorkerPipeSessionBackgroundTaskStopTimedOut") + && entry.Fields.TryGetValue("task", out object? task) + && (task as string) == "PipeRead"); + } + + /// + /// WRK-31. The session now owns and closes the transport rather than leaving it to + /// WorkerPipeClient's using, because the read it unblocks has to be observed + /// while the session still holds it. This asserts the closure actually happens on the + /// session's own timeline: the gateway end of the pipe must see disconnection while + /// is still undisposed, so nothing but the worker side of the + /// session can have closed it. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_WhenSessionEnds_ClosesTransportBeforeTheHarnessDoes() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + + // PipePair.Dispose has not run — its `using` is still in scope — so the only thing that can + // have closed the worker end is the session. A gateway-side read is uncancellable on net48 + // exactly as the worker's is, so a session that left the pipe open would park this read + // until the harness disposes; the bound below is what catches that. + Task disconnectTask = ReadUntilDisconnectedAsync(pipePair.GatewayReader); + Task completedTask = await Task.WhenAny( + disconnectTask, + Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token)); + Assert.Same(disconnectTask, completedTask); + + Exception disconnect = await disconnectTask; + Assert.True( + disconnect is IOException + || disconnect is ObjectDisposedException + || (disconnect is WorkerFrameProtocolException frameException + && frameException.ErrorCode == WorkerFrameProtocolErrorCode.EndOfStream), + $"Expected the gateway read to observe a pipe disconnection, got {disconnect}."); + } + /// /// Verifies that ShutdownWorker returns its OK reply BEFORE the graceful /// shutdown runs and disposes the runtime session, and that the message @@ -1881,7 +2067,11 @@ public sealed class WorkerPipeSessionTests () => 1234, sessionOptions, () => runtime, - logger); + logger, + // Hand the session the same ownership the production WorkerPipeClient path gives it, so + // these tests exercise the real teardown: the session closes the transport itself and + // then observes the read that closure unblocks. + transportStream: stream); } private static WorkerFrameProtocolOptions CreateOptions() @@ -2149,6 +2339,32 @@ public sealed class WorkerPipeSessionTests } } + /// + /// Reads a pipe end until it stops producing frames, returning whatever ended it. Frames + /// still buffered from before the peer closed its handle — a trailing heartbeat, say — are + /// drained first, because Windows named pipes hand over buffered bytes ahead of the + /// broken-pipe signal. + /// + /// Frame reader over the end being watched. + /// The exception that ended the read. + private static async Task ReadUntilDisconnectedAsync(WorkerFrameReader reader) + { + while (true) + { + try + { + await reader.ReadAsync(CancellationToken.None).ConfigureAwait(false); + } + catch (Exception exception) when ( + exception is IOException + || exception is ObjectDisposedException + || exception is WorkerFrameProtocolException) + { + return exception; + } + } + } + private static WorkerEnvelope[] ReadWrittenFrames( MemoryStream stream, WorkerFrameProtocolOptions options) diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs index 041e879..8ae6601 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs @@ -22,6 +22,14 @@ public sealed class WorkerFrameReader // Reused across frames rather than allocated per read (GWC-30). Safe because ReadAsync is // single-consumer by construction; the prefix is fully overwritten by every read. + // + // The single-consumer invariant has to survive teardown as well as steady state, because a read + // abandoned by WorkerPipeSession's message loop still owns this buffer (and its rented payload + // buffer) until it faults. Nothing may call ReadAsync again after that point: a second read + // would race the abandoned one for the prefix, and could hand a pooled payload buffer back to + // ArrayPool twice. The loop guarantees it structurally — it only ever issues a read after + // awaiting the previous one, and it never re-enters after unwinding — and teardown only awaits + // the abandoned read, never reissues it (WorkerPipeSession.ObserveAbandonedPipeReadAsync). private readonly byte[] _lengthPrefix = new byte[sizeof(uint)]; /// Initializes the reader with a stream and protocol options. @@ -106,6 +114,12 @@ public sealed class WorkerFrameReader int offset = 0; while (offset < count) { + // The token is forwarded but is NOT a bound on a pipe read: on .NET Framework 4.8 + // NamedPipeClientStream.ReadAsync accepts a CancellationToken and never wires it to the + // overlapped I/O, so a read waiting on gateway bytes ignores cancellation entirely. Only + // closing the handle ends it (WorkerPipeSession disposes the transport at teardown for + // exactly this reason). It is still passed because non-pipe streams — the + // MemoryStream-backed unit tests, and any future transport — do honor it. int bytesRead = await _stream .ReadAsync(buffer, offset, count - offset, cancellationToken) .ConfigureAwait(false); diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeClient.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeClient.cs index ab47518..03885cf 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeClient.cs @@ -151,6 +151,13 @@ public sealed class WorkerPipeClient : IWorkerPipeClient WorkerFrameProtocolOptions frameOptions = new(options); + // The session disposes this pipe itself as its last teardown step — that disposal is what + // unblocks a net48 pipe read the message loop abandoned, and it has to happen while the + // session still holds the read task so the resulting fault is observed rather than orphaned + // (see WorkerPipeSession.RunAsync). The `using` stays as the backstop for the paths the + // session never reaches: a session factory that throws, or a RunAsync that never gets past + // its own construction. Disposal is idempotent, so the second Dispose is a no-op — do not + // "clean up" this `using` on the assumption that it is now redundant. using NamedPipeClientStream pipe = await ConnectWithRetryAsync(options.PipeName, cancellationToken) .ConfigureAwait(false); diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index 8934946..7f1eb3e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -39,8 +39,21 @@ public sealed class WorkerPipeSession private readonly WorkerFrameWriter _writer; private readonly object _commandTaskGate = new(); private readonly HashSet _activeCommandTasks = new(); + + // The transport the reader and writer share, when this session was handed one to own. Null for + // the reader/writer constructors, whose callers keep ownership of whatever streams they built the + // pair over. Owning it is what lets teardown close the handle (WRK-31): on net48 a pipe read + // parked in the kernel cannot be cancelled, only unblocked by disposal. + private readonly Stream? _transportStream; + private IWorkerRuntimeSession? _runtimeSession; + // The one outstanding, not-yet-awaited frame read, or null when no read is in flight. Written + // only by the message loop (which sets it at each read issue and clears it before awaiting that + // read itself) and read only by RunAsync's finally — which runs after the loop's task has + // completed, so the await supplies the happens-before edge and no interlock is needed. + private Task? _pendingReadTask; + // Mutated from the message loop, command tasks, the heartbeat loop and the // shutdown path; volatile so cross-thread reads observe the latest state // without tearing (WorkerState is an int-backed protobuf enum). @@ -63,7 +76,8 @@ public sealed class WorkerPipeSession () => Process.GetCurrentProcess().Id, new WorkerPipeSessionOptions(), () => new MxAccessStaSession((eq, affinity, comFactory) => new AlarmCommandHandler(eq, () => new WnWrapAlarmConsumer(), affinity, comFactory, standbyFactory: null)), - logger) + logger, + stream) { } @@ -96,6 +110,12 @@ public sealed class WorkerPipeSession /// Session-specific options. /// Factory creating the MXAccess runtime session. /// Optional logger for diagnostic output. + /// + /// Stream the reader and writer share, when this session is to own it. Supplying it makes + /// dispose the transport as its last teardown step, which is the only + /// way to unblock a pending net48 pipe read (see ). Null + /// leaves ownership — and the disposal — with the caller. + /// public WorkerPipeSession( WorkerFrameReader reader, WorkerFrameWriter writer, @@ -103,7 +123,8 @@ public sealed class WorkerPipeSession Func processIdProvider, WorkerPipeSessionOptions sessionOptions, Func runtimeSessionFactory, - IWorkerLogger? logger = null) + IWorkerLogger? logger = null, + Stream? transportStream = null) { _reader = reader ?? throw new ArgumentNullException(nameof(reader)); _writer = writer ?? throw new ArgumentNullException(nameof(writer)); @@ -112,6 +133,7 @@ public sealed class WorkerPipeSession _sessionOptions = sessionOptions ?? throw new ArgumentNullException(nameof(sessionOptions)); _runtimeSessionFactory = runtimeSessionFactory ?? throw new ArgumentNullException(nameof(runtimeSessionFactory)); _logger = logger; + _transportStream = transportStream; _sessionOptions.Validate(); } @@ -142,9 +164,92 @@ public sealed class WorkerPipeSession _runtimeSession?.Dispose(); _runtimeSession = null; _state = WorkerState.Stopped; + + // 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. + // + // 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 + // task this session still holds, so ObserveAbandonedPipeReadAsync can await it instead of + // leaving it for a TaskScheduler.UnobservedTaskException handler the worker does not have. + DisposeTransportStream(); + await ObserveAbandonedPipeReadAsync().ConfigureAwait(false); } } + /// + /// 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. + /// + private void DisposeTransportStream() + { + if (_transportStream is null) + { + return; + } + + try + { + _transportStream.Dispose(); + } + catch (Exception exception) when (exception is IOException || exception is ObjectDisposedException) + { + _logger?.Error( + "WorkerPipeSessionTransportDisposeFailed", + new Dictionary + { + ["session_id"] = _options.SessionId, + ["exception"] = exception.ToString(), + }); + } + } + + /// + /// Awaits the frame read the message loop walked away from, if there is one. + /// races an uncancellable read against the heartbeat and + /// 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 + /// 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. + /// + /// A task that represents the asynchronous operation. + private async Task ObserveAbandonedPipeReadAsync() + { + Task? readTask = _pendingReadTask; + _pendingReadTask = null; + if (readTask is null) + { + return; + } + + await ObserveBackgroundTaskStopAsync(readTask, "PipeRead").ConfigureAwait(false); + } + /// Completes the gateway startup handshake using default MXAccess initialization. /// Token to cancel the asynchronous operation. /// A task that represents the asynchronous operation. @@ -264,6 +369,34 @@ public sealed class WorkerPipeSession return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken); } + /// + /// Runs the post-handshake message loop, racing one outstanding frame read against the + /// heartbeat and event-drain loops. + /// + /// + /// + /// loopCancellation does NOT bound the read. On .NET Framework 4.8 + /// NamedPipeClientStream.ReadAsync accepts a and + /// then ignores it — the token never reaches the overlapped I/O, so a read parked + /// waiting for gateway bytes stays parked no matter what is cancelled. The token is + /// still passed because the reader's contract takes one and a non-pipe stream (the + /// MemoryStream-backed unit tests) does honor it. What actually ends a parked read is + /// closing the handle, which does at the end of teardown. + /// + /// + /// That asymmetry is why the loop records its outstanding read in + /// _pendingReadTask. Every fault exit — an event-drain fault, an event too large + /// to frame, a failed heartbeat write — unwinds through Task.WhenAny while the + /// read is still pending, and the fault that read eventually takes has to be observed by + /// somebody (see ). The field is cleared + /// before the loop awaits a read itself, so it is non-null exactly when a read is + /// outstanding and unobserved. Graceful exits — the return below, after a + /// WorkerShutdown envelope or a ShutdownWorker command — leave no pending + /// read at all, so teardown's observation is a no-op there. + /// + /// + /// Token to cancel the asynchronous operation. + /// A task that represents the asynchronous operation. private async Task RunMessageLoopAsync(CancellationToken cancellationToken) { using CancellationTokenSource loopCancellation = CancellationTokenSource @@ -273,6 +406,7 @@ public sealed class WorkerPipeSession Task heartbeatTask = RunHeartbeatLoopAsync(heartbeatCancellation.Token); Task eventDrainTask = RunEventDrainLoopAsync(heartbeatCancellation.Token); Task readTask = _reader.ReadAsync(loopCancellation.Token); + _pendingReadTask = readTask; try { @@ -281,6 +415,9 @@ public sealed class WorkerPipeSession Task completedTask = await Task.WhenAny(readTask, heartbeatTask, eventDrainTask).ConfigureAwait(false); if (completedTask == readTask) { + // The loop observes this read itself, whether it yields an envelope or throws, + // so it is no longer the abandoned one teardown has to account for. + _pendingReadTask = null; WorkerEnvelope envelope = await readTask.ConfigureAwait(false); bool keepReading = await DispatchGatewayEnvelopeAsync(envelope, cancellationToken).ConfigureAwait(false); if (!keepReading) @@ -289,6 +426,7 @@ public sealed class WorkerPipeSession } readTask = _reader.ReadAsync(loopCancellation.Token); + _pendingReadTask = readTask; } else if (completedTask == heartbeatTask) {