From cc1e895474afef5d12b9a937566d4319e90776c9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:48:31 -0400 Subject: [PATCH] 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. --- docs/Sessions.md | 6 + .../Sessions/GatewaySession.cs | 52 +++++- .../Sessions/GatewaySessionWorkerExitTests.cs | 151 ++++++++++++++++++ 3 files changed, 204 insertions(+), 5 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs diff --git a/docs/Sessions.md b/docs/Sessions.md index b945fd0..0ac66a0 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -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. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index 8cdb36f..5d57f0c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -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)."); } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs new file mode 100644 index 0000000..2a17c8b --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs @@ -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; + +/// +/// A worker that goes away on its own — the client sent the ShutdownWorker control +/// command and the worker exited cleanly, or the worker crashed / its pipe broke — must be +/// noticed by its . The real reaches +/// Closed / Faulted, but nothing used to tell the session, which kept reporting +/// Ready (live windev finding) until its lease expired. +/// +public sealed class GatewaySessionWorkerExitTests +{ + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5); + + /// + /// After the client's ShutdownWorker 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. + /// + /// A task that represents the asynchronous operation. + [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 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); + } + + /// + /// The other way a worker goes away: its pipe breaks (crash). The worker client faults, and + /// the session must follow it to Faulted rather than stay Ready. + /// + /// A task that represents the asynchronous operation. + [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 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 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 + { + /// + public Task CreateAsync(GatewaySession session, CancellationToken cancellationToken) + { + return Task.FromResult(client); + } + } +}