fix(worker): session-owned stream disposal unblocks and observes the net48 pipe read at teardown

src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs

@@ -912,6 +912,192 @@ 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
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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");
+    }
+
+    /// <summary>
+    ///     WRK-31. The session now owns and closes the transport rather than leaving it to
+    ///     <c>WorkerPipeClient</c>'s <c>using</c>, 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
+    ///     <see cref="PipePair"/> is still undisposed, so nothing but the worker side of the
+    ///     session can have closed it.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<Exception> 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}.");
+    }
+
     /// <summary>
     ///     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
         }
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <param name="reader">Frame reader over the end being watched.</param>
+    /// <returns>The exception that ended the read.</returns>
+    private static async Task<Exception> 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)