fix(sessions): fault a Ready session whose worker reached a terminal state on its own

Live windev finding: after a client sent the ShutdownWorker control command the worker exited 0 and WorkerClient moved to Closed on the shutdown ack, but GatewaySession.State stayed Ready (even with an event stream attached) until its lease expired. IWorkerClient has no state-change callback, so the session observes the worker from its own worker-event pump source: WorkerClient enters Closed/Faulted before completing the event channel, so when the enumeration ends ObserveWorkerTerminalState sees the terminal state and faults a Ready session. Faulted rather than Closed: Closed means the gateway closed it and would linger in the registry; a faulted session stays observable with the reason and the faulted-session sweep reaps it, accounting SessionClosed exactly once. Only a Ready session follows its worker, so a gateway- initiated close is unaffected. FakeWorkerHarness-based tests cover the ShutdownWorker exit and a broken worker pipe.
2026-10-22 06:48:31 -04:00
parent ec9f26400a
commit cc1e895474
3 changed files with 204 additions and 5 deletions
@@ -247,6 +247,12 @@ Mechanically: when the last external subscriber detaches and `DetachGraceSeconds

 `DetachGraceSeconds` controls retention and expiry only; the reconnect/replay path that re-attaches a dropped client to a retained session is described in [Reconnect and replay](#reconnect-and-replay).

+#### Worker exit outside a gateway close
+
+`IWorkerClient` has no state-change callback, so a worker that reached `Closed` or `Faulted` *without the gateway closing it* — the client sent the `ShutdownWorker` control command and the worker exited cleanly, the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session reporting `Ready` until its lease expired (a live finding: `WorkerClient` was `Closed` on the shutdown ack, the session still `Ready`, even with an event stream attached). The session now observes it from its own worker-event pump source (`MapWorkerEventsAsync`): the single distributor pump is the session's one always-on view of the worker (it runs from `MarkReady` via the dashboard mirror), and `WorkerClient` enters its terminal state *before* it completes the event channel, so when the enumeration ends the terminal state is visible. `ObserveWorkerTerminalState` then faults a `Ready` session whose worker is `Closed`/`Faulted`.
+
+The choice is **`Faulted`, not `Closed`**: `Closed` means the gateway closed the session, and a `Closed` session left in the registry would linger until its lease expired with nothing reaping it. A faulted session is permanently unusable (every command fails the readiness check, as the worker is gone), stays observable via `GetSessionStatus` with `FinalFault` naming the worker state, and is reaped by the faulted-session sweep below after `FaultedGraceSeconds` — which is where `SessionClosed` is accounted, exactly once, through the ordinary `CloseSessionCoreAsync` path (`WorkerClient.ShutdownAsync` on a `Closed` client returns immediately, so no kill fallback or warning). Only a `Ready` session follows its worker: during a gateway-initiated close the worker closing is the expected outcome, not a fault.
+
 #### Faulted-session reaping

 A session that faults (for example, a slow single-subscriber client that overflows its event queue under the `FailFast` backpressure policy) is left permanently unusable — every command fails `EvaluateReadyUnderLock` — but its worker keeps running and it still pins one of `MxGateway:Sessions:MaxSessions` slots. Rather than waiting up to `DefaultLeaseSeconds` for the lease to expire, `SessionManager.CloseExpiredLeasesAsync` also checks `IsFaultedReapable(now)` and reaps a faulted session through the same `TryBeginCloseIfExpired` → `CloseSessionCoreAsync` teardown, with the distinct `FaultedReason` (`"faulted-reaped"`). `MarkFaulted` stamps a fault timestamp from the session's `TimeProvider`; `MxGateway:Sessions:FaultedGraceSeconds` (default `0`) bounds how long the faulted session is retained before the next sweep reaps it — `0` reaps it on the next sweep, a positive value keeps it observable via `GetSessionStatus` for that window first. Sweep precedence when several conditions fire at once is lease-expiry, then faulted, then detach-grace, then unattached-grace.
@@ -1004,12 +1004,54 @@ public sealed class GatewaySession
        IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
        TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow());

-        await foreach (WorkerEvent workerEvent in workerClient
-            .ReadEventsAsync(cancellationToken)
-            .WithCancellation(cancellationToken)
-            .ConfigureAwait(false))
+        try
        {
-            yield return mapper.MapEvent(workerEvent);
+            await foreach (WorkerEvent workerEvent in workerClient
+                .ReadEventsAsync(cancellationToken)
+                .WithCancellation(cancellationToken)
+                .ConfigureAwait(false))
+            {
+                yield return mapper.MapEvent(workerEvent);
+            }
+        }
+        finally
+        {
+            // The worker event stream only ends when the worker client has reached a terminal
+            // state (it enters Closed/Faulted BEFORE completing the channel) or the pump is being
+            // torn down. This is the session's one always-on view of the worker, so it is where
+            // a worker that went away on its own is noticed.
+            ObserveWorkerTerminalState();
+        }
+    }
+
+    // IWorkerClient has no state-change callback, so a worker that reached Closed/Faulted
+    // without the gateway closing it — the client sent ShutdownWorker and the worker exited,
+    // the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session
+    // reporting Ready until its lease expired (live finding). The session follows the worker
+    // into Faulted: it is permanently unusable (every command fails the readiness check), stays
+    // observable via GetSessionStatus with the reason, and the faulted-session sweep reaps it
+    // after FaultedGraceSeconds, accounting SessionClosed exactly once there. Not Closed: Closed
+    // means the gateway closed it, and a Closed session still in the registry would linger until
+    // its lease expired with nothing reaping it. Only a Ready session follows — during a
+    // gateway-initiated close the worker closing is the expected outcome, not a fault.
+    private void ObserveWorkerTerminalState()
+    {
+        lock (_syncRoot)
+        {
+            if (_state is not SessionState.Ready || _workerClient is null)
+            {
+                return;
+            }
+
+            WorkerClientState workerState = _workerClient.State;
+            if (workerState is not (WorkerClientState.Closed or WorkerClientState.Faulted))
+            {
+                return;
+            }
+
+            MarkFaulted(
+                $"Worker for session {SessionId} reached {workerState} outside a gateway-initiated close "
+                + "(worker exited or faulted).");
        }
    }

@@ -0,0 +1,151 @@
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Sessions;
+
+/// <summary>
+///     A worker that goes away on its own — the client sent the <c>ShutdownWorker</c> control
+///     command and the worker exited cleanly, or the worker crashed / its pipe broke — must be
+///     noticed by its <see cref="GatewaySession"/>. The real <see cref="WorkerClient"/> reaches
+///     <c>Closed</c> / <c>Faulted</c>, but nothing used to tell the session, which kept reporting
+///     <c>Ready</c> (live windev finding) until its lease expired.
+/// </summary>
+public sealed class GatewaySessionWorkerExitTests
+{
+    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
+
+    /// <summary>
+    ///     After the client's <c>ShutdownWorker</c> command the fake worker replies OK, sends the
+    ///     shutdown ack and the worker client closes. The session must fault (the worker is gone
+    ///     outside a gateway-initiated close), stay observable with a fault reason, and then be
+    ///     reaped by the ordinary faulted-session sweep — which is where the session is accounted
+    ///     closed, exactly once.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ClientShutdownWorker_WorkerExits_SessionFaultsAndSweepReapsItOnce()
+    {
+        await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
+        using GatewayMetrics metrics = new();
+        WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
+        await StartClientAsync(fakeWorker, client);
+        SessionManager manager = CreateManager(client, metrics);
+        GatewaySession session = await OpenSessionAsync(manager);
+        using IEventSubscriberLease stream = session.AttachEventSubscriber(maxSubscribers: 1);
+
+        Task<WorkerCommandReply> invokeTask = manager.InvokeAsync(session.SessionId, CreateShutdownWorkerCommand(), CancellationToken.None);
+        await fakeWorker.RespondToControlCommandAsync().WaitAsync(TestTimeout);
+        WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout);
+        Assert.Equal(ProtocolStatusCode.Ok, reply.Reply.ProtocolStatus.Code);
+
+        await WaitUntilAsync(() => session.State == SessionState.Faulted);
+
+        Assert.Equal(WorkerClientState.Closed, client.State);
+        Assert.Contains("worker", session.FinalFault, StringComparison.OrdinalIgnoreCase);
+        Assert.True(manager.TryGetSession(session.SessionId, out _));
+        Assert.Equal(1, metrics.GetSnapshot().OpenSessions);
+        Assert.Equal(0, metrics.GetSnapshot().SessionsClosed);
+
+        int closedCount = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
+
+        Assert.Equal(1, closedCount);
+        Assert.Equal(SessionState.Closed, session.State);
+        Assert.False(manager.TryGetSession(session.SessionId, out _));
+        GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
+        Assert.Equal(1, snapshot.SessionsClosed);
+        Assert.Equal(0, snapshot.OpenSessions);
+    }
+
+    /// <summary>
+    ///     The other way a worker goes away: its pipe breaks (crash). The worker client faults, and
+    ///     the session must follow it to <c>Faulted</c> rather than stay <c>Ready</c>.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WorkerPipeBreaks_SessionFaults()
+    {
+        await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
+        using GatewayMetrics metrics = new();
+        WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
+        await StartClientAsync(fakeWorker, client);
+        SessionManager manager = CreateManager(client, metrics);
+        GatewaySession session = await OpenSessionAsync(manager);
+        using IEventSubscriberLease stream = session.AttachEventSubscriber(maxSubscribers: 1);
+
+        await fakeWorker.DisposeWorkerSideAsync();
+
+        await WaitUntilAsync(() => session.State == SessionState.Faulted);
+        Assert.Equal(WorkerClientState.Faulted, client.State);
+
+        await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
+        Assert.False(manager.TryGetSession(session.SessionId, out _));
+        Assert.Equal(1, metrics.GetSnapshot().SessionsClosed);
+        Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
+    }
+
+    private static async Task StartClientAsync(FakeWorkerHarness fakeWorker, WorkerClient client)
+    {
+        Task startTask = client.StartAsync(CancellationToken.None);
+        await fakeWorker.CompleteStartupAsync();
+        await startTask.WaitAsync(TestTimeout);
+        Assert.Equal(WorkerClientState.Ready, client.State);
+    }
+
+    private static SessionManager CreateManager(IWorkerClient client, GatewayMetrics metrics)
+    {
+        return new SessionManager(
+            new SessionRegistry(),
+            new SingleClientFactory(client),
+            Options.Create(new GatewayOptions()),
+            metrics);
+    }
+
+    private static Task<GatewaySession> OpenSessionAsync(SessionManager manager)
+    {
+        return manager.OpenSessionAsync(
+            new SessionOpenRequest(RequestedBackend: null, ClientSessionName: "worker-exit", ClientCorrelationId: null, CommandTimeout: null),
+            "client-1",
+            ownerKeyId: null,
+            CancellationToken.None);
+    }
+
+    private static WorkerCommand CreateShutdownWorkerCommand()
+    {
+        return new WorkerCommand
+        {
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.ShutdownWorker,
+                ShutdownWorker = new ShutdownWorkerCommand(),
+            },
+        };
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition)
+    {
+        DateTime deadline = DateTime.UtcNow + TestTimeout;
+        while (!condition())
+        {
+            if (DateTime.UtcNow > deadline)
+            {
+                Assert.Fail("Condition was not met within the test timeout.");
+            }
+
+            await Task.Delay(10);
+        }
+    }
+
+    private sealed class SingleClientFactory(IWorkerClient client) : ISessionWorkerClientFactory
+    {
+        /// <inheritdoc />
+        public Task<IWorkerClient> CreateAsync(GatewaySession session, CancellationToken cancellationToken)
+        {
+            return Task.FromResult(client);
+        }
+    }
+}