fix(sessions): refuse a dead worker at attach, fault on pump readiness failure, keep cancel/stopping out of OpenFailed

- TryAttachWorkerClientAndMarkReadyAsync returns SessionAttachOutcome and refuses a
  worker client that is already Faulted/Closed; the open discards it and fails OpenFailed.
- MapWorkerEventsAsync calls ObserveWorkerTerminalState when its first readiness check
  fails, so a worker that dies between attach and the first pump read faults the session
  and the faulted sweep reaps it.
- A caller-cancelled open rolls back, records no fault and rethrows the cancellation
  (gRPC Cancelled); an attach refused because the drain closed the session reports
  GatewayStopping.
- docs/Sessions.md updated.
This commit is contained in:
Joseph Doherty
2026-10-22 08:41:02 -04:00
parent 2375df5df3
commit a178d68b60
6 changed files with 256 additions and 31 deletions
@@ -475,17 +475,27 @@ public sealed class GatewaySession
/// accounting here, and <see cref="OpenRecorded"/> is set right after it returns.
/// </param>
/// <returns>
/// <see langword="true"/> when the session became Ready; <see langword="false"/> when the
/// attach was refused (see <see cref="AttachWorkerClient"/>), including when the session was
/// disposed before the gate could be taken.
/// <see cref="SessionAttachOutcome.Attached"/> when the session became Ready;
/// <see cref="SessionAttachOutcome.SessionClosed"/> when the session refused the attach
/// (see <see cref="AttachWorkerClient"/>), including when it was disposed before the gate
/// could be taken; <see cref="SessionAttachOutcome.WorkerTerminal"/> when the worker client
/// had already reached <see cref="WorkerClientState.Faulted"/> or
/// <see cref="WorkerClientState.Closed"/>.
/// </returns>
/// <remarks>
/// Not cancellable: the worker client already exists, so the only acceptable outcomes are
/// "attached" or "refused, caller discards it". The gate is held only for the duration of
/// a close or kill, so the wait is bounded by the worker shutdown timeout. The caller
/// still decides what to do about a cancelled open before calling this.
/// <para>
/// A worker that dies right after its handshake (an STA loop fault, exit code 7) can
/// be terminal before it is attached. Attaching it would hand the caller a Ready
/// session over a dead worker that nothing notices until the lease expires, so it is
/// refused here. A worker that dies just after this check is caught by the event
/// pump's readiness failure path instead (see <c>MapWorkerEventsAsync</c>).
/// </para>
/// </remarks>
internal async ValueTask<bool> TryAttachWorkerClientAndMarkReadyAsync(
internal async ValueTask<SessionAttachOutcome> TryAttachWorkerClientAndMarkReadyAsync(
IWorkerClient workerClient,
Action onReady)
{
@@ -499,14 +509,27 @@ public sealed class GatewaySession
catch (ObjectDisposedException)
{
// Disposed while the worker was starting: the session has been torn down.
return false;
return SessionAttachOutcome.SessionClosed;
}
try
{
if (workerClient.State is WorkerClientState.Faulted or WorkerClientState.Closed)
{
// A session closed meanwhile is the stronger reason: it decides whether the open
// reports a stopping gateway rather than a failed worker.
lock (_syncRoot)
{
return _state is SessionState.Closing or SessionState.Closed or SessionState.Faulted
|| _eventDistributorDisposed
? SessionAttachOutcome.SessionClosed
: SessionAttachOutcome.WorkerTerminal;
}
}
if (!AttachWorkerClient(workerClient))
{
return false;
return SessionAttachOutcome.SessionClosed;
}
MarkReady();
@@ -516,7 +539,7 @@ public sealed class GatewaySession
_openRecorded = true;
}
return true;
return SessionAttachOutcome.Attached;
}
finally
{
@@ -1001,7 +1024,22 @@ public sealed class GatewaySession
[EnumeratorCancellation] CancellationToken cancellationToken)
{
MxAccessGrpcMapper mapper = _eventStreaming.Mapper;
IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
IWorkerClient workerClient;
try
{
workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
}
catch (SessionManagerException)
{
// The worker can die between the open's attach and this first read (an STA loop
// fault or exit code 7 right after the handshake). The readiness check then fails
// here and the worker event stream below is never reached, so its finally never
// runs; without this the session would stay Ready over a dead worker until its
// lease expired.
ObserveWorkerTerminalState();
throw;
}
TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow());
try
@@ -0,0 +1,26 @@
namespace ZB.MOM.WW.MxGateway.Server.Sessions;
/// <summary>
/// Result of <see cref="GatewaySession.TryAttachWorkerClientAndMarkReadyAsync"/>: whether the
/// open's worker client was attached, and if not, which side refused it. The open maps each
/// refusal to a different failure, so the reason has to survive the call.
/// </summary>
internal enum SessionAttachOutcome
{
/// <summary>The worker client was attached and the session is <see cref="SessionState.Ready"/>.</summary>
Attached,
/// <summary>
/// The session was closed, faulted or disposed while its worker was starting (an admin
/// kill or the host drain); the caller discards the client.
/// </summary>
SessionClosed,
/// <summary>
/// The worker client was already <see cref="Workers.WorkerClientState.Faulted"/> or
/// <see cref="Workers.WorkerClientState.Closed"/> — the worker died right after its
/// handshake. Attaching it would leave a Ready session over a dead worker; the caller
/// discards the client and fails the open.
/// </summary>
WorkerTerminal,
}
@@ -164,11 +164,11 @@ public sealed class SessionManager : ISessionManager
// if a kill or the host drain closed it meanwhile — it must not be returned on a dead
// session; and a caller that abandoned the open while the worker was starting gets
// no session either.
bool ready;
SessionAttachOutcome outcome;
try
{
cancellationToken.ThrowIfCancellationRequested();
ready = await session
outcome = await session
.TryAttachWorkerClientAndMarkReadyAsync(workerClient, _metrics.SessionOpened)
.ConfigureAwait(false);
}
@@ -178,14 +178,25 @@ public sealed class SessionManager : ISessionManager
throw;
}
if (!ready)
if (outcome is SessionAttachOutcome.Attached)
{
await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false);
throw new InvalidOperationException(
$"Session {session.SessionId} was closed while its worker was starting.");
return session;
}
return session;
WorkerClientState refusedWorkerState = workerClient.State;
await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false);
if (outcome is SessionAttachOutcome.WorkerTerminal)
{
throw new InvalidOperationException(
$"The worker for session {session.SessionId} reached {refusedWorkerState} before it could be attached.");
}
// Refused because the session was closed while its worker was starting. When the host
// drain did that, the caller should retry against the restarted gateway, exactly like
// an open refused at the door, rather than see an open failure.
ThrowIfStopping();
throw new InvalidOperationException(
$"Session {session.SessionId} was closed while its worker was starting.");
}
catch (SessionManagerException exception) when (exception.ErrorCode == SessionManagerErrorCode.GatewayStopping)
{
@@ -194,6 +205,15 @@ public sealed class SessionManager : ISessionManager
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
throw;
}
catch (OperationCanceledException exception) when (cancellationToken.IsCancellationRequested)
{
// The caller abandoned the open (deadline, disconnect). Not an open failure either:
// roll back (a worker that arrived late was already discarded above), record no
// fault, log nothing, and let the cancellation surface so the gRPC layer maps it to
// Cancelled rather than Unavailable.
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
throw;
}
catch (Exception exception)
{
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
@@ -88,6 +88,78 @@ public sealed class GatewaySessionWorkerExitTests
Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
}
/// <summary>
/// A worker that dies right after its handshake (STA loop fault, exit code 7) can reach
/// <c>Faulted</c> before the open attaches it. The attach must refuse such a client: the
/// open fails with <c>OpenFailed</c>, the client is discarded, and no session is left
/// registered reporting <c>Ready</c> over a dead worker until its lease expires.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task OpenSession_WorkerAlreadyFaultedAtAttach_FailsOpenAndDiscardsClient()
{
await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
using GatewayMetrics metrics = new();
WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
await StartClientAsync(fakeWorker, client);
await fakeWorker.DisposeWorkerSideAsync();
await WaitUntilAsync(() => client.State == WorkerClientState.Faulted);
SessionRegistry registry = new();
SessionManager manager = new(
registry,
new SingleClientFactory(client),
Options.Create(new GatewayOptions()),
metrics);
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
async () => await OpenSessionAsync(manager));
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
Assert.Contains("Faulted", exception.InnerException?.Message, StringComparison.Ordinal);
Assert.Equal(0, registry.Count);
GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
Assert.Equal(0, snapshot.OpenSessions);
Assert.Equal(0, snapshot.SessionsOpened);
}
/// <summary>
/// The worker is still <c>Ready</c> when the open attaches it but faults before the event
/// pump's first read (which the dashboard mirror starts at <c>MarkReady</c>). The pump's
/// readiness check then fails without ever reaching the worker event stream, so it is that
/// failure path that must notice the dead worker: the session follows it to
/// <c>Faulted</c> and the faulted-session sweep reaps it, accounting the close once.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WorkerFaultsBetweenAttachAndFirstPumpRead_SessionFaultsAndSweepReapsItOnce()
{
using GatewayMetrics metrics = new();
ReadyOnceWorkerClient client = new();
SessionManager manager = new(
new SessionRegistry(),
new SingleClientFactory(client),
Options.Create(new GatewayOptions()),
metrics,
dashboardEventBroadcaster: TestSupport.NullDashboardEventBroadcaster.Instance);
GatewaySession session = await OpenSessionAsync(manager);
await WaitUntilAsync(() => session.State == SessionState.Faulted);
Assert.Contains("worker", session.FinalFault, StringComparison.OrdinalIgnoreCase);
Assert.Equal(1, metrics.GetSnapshot().OpenSessions);
int closedCount = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
int secondPass = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
Assert.Equal(1, closedCount);
Assert.Equal(0, secondPass);
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);
}
private static async Task StartClientAsync(FakeWorkerHarness fakeWorker, WorkerClient client)
{
Task startTask = client.StartAsync(CancellationToken.None);
@@ -148,4 +220,56 @@ public sealed class GatewaySessionWorkerExitTests
return Task.FromResult(client);
}
}
/// <summary>
/// Worker client that reports <c>Ready</c> for its first state read (the open's attach
/// check) and <c>Faulted</c> from then on, reproducing a worker that dies in the window
/// between the attach and the event pump's first readiness check.
/// </summary>
private sealed class ReadyOnceWorkerClient : IWorkerClient
{
private int _stateReads;
/// <inheritdoc />
public string SessionId => "session-ready-once";
/// <inheritdoc />
public int? ProcessId => 4321;
/// <inheritdoc />
public WorkerClientState State
=> Interlocked.Increment(ref _stateReads) == 1 ? WorkerClientState.Ready : WorkerClientState.Faulted;
/// <inheritdoc />
public DateTimeOffset LastHeartbeatAt => DateTimeOffset.UtcNow;
/// <inheritdoc />
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(WorkerCommand command, TimeSpan timeout, CancellationToken cancellationToken)
=> throw new InvalidOperationException("The worker is faulted.");
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
await Task.CompletedTask;
throw new InvalidOperationException("The pump must not reach the event stream of a faulted worker.");
#pragma warning disable CS0162 // Unreachable code: the yield makes this an async iterator.
yield break;
#pragma warning restore CS0162
}
/// <inheritdoc />
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
/// <inheritdoc />
public void Kill(string reason)
{
}
/// <inheritdoc />
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
}
@@ -533,13 +533,17 @@ public sealed class SessionManagerTests
// which flaked under CI load) so a regression that waited out the timeout is caught
// while ordinary scheduling jitter never trips it.
const int readyWaitTimeoutMs = 5000;
FakeWorkerClient workerClient = new() { State = WorkerClientState.Faulted };
FakeWorkerClient workerClient = new();
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(workerReadyWaitTimeoutMs: readyWaitTimeoutMs));
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
Assert.Equal(SessionState.Ready, session.State);
// The worker faults after the open attached it (an open refuses an already-faulted
// worker). No event pump runs here, so nothing has moved the session off Ready yet.
workerClient.State = WorkerClientState.Faulted;
Stopwatch stopwatch = Stopwatch.StartNew();
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
async () => await manager.InvokeAsync(
@@ -1145,7 +1149,9 @@ public sealed class SessionManagerTests
Assert.Equal(0, snapshot.SessionsClosed);
// The slot the killed open held must have been released exactly once: a replacement open
// succeeds against MaxSessions = 1 (the released gate hands the same fake straight back).
// succeeds against MaxSessions = 1 (the released gate hands the same fake straight back,
// so undo the discard's kill first — an open refuses an already-faulted worker).
workerClient.State = WorkerClientState.Ready;
GatewaySession reopened = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
Assert.Equal(SessionState.Ready, reopened.State);
Assert.Equal(1, registry.Count);
@@ -1155,11 +1161,13 @@ public sealed class SessionManagerTests
/// A caller that abandons its open (deadline, disconnect) while the worker is starting
/// must not leave the worker running: whatever fails between the worker arriving and the
/// session becoming Ready — here the caller's own cancellation — the late worker client is
/// killed and disposed before the failure surfaces.
/// killed and disposed before the failure surfaces. The cancellation is not an open
/// failure: it surfaces as the <see cref="OperationCanceledException"/> itself (which the
/// gRPC layer maps to <c>Cancelled</c>) and records no fault.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task OpenSessionAsync_WhenCallerCancelledDuringWorkerStartup_DiscardsWorkerClient()
public async Task OpenSessionAsync_WhenCallerCancelledDuringWorkerStartup_DiscardsWorkerClientAndRethrowsCancellation()
{
FakeWorkerClient workerClient = new();
GatedSessionWorkerClientFactory factory = new(workerClient);
@@ -1173,10 +1181,9 @@ public sealed class SessionManagerTests
await cancellation.CancelAsync();
factory.ReleaseCreate();
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () => await openTask);
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await openTask);
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
Assert.IsAssignableFrom<OperationCanceledException>(exception.InnerException);
Assert.Equal(0, metrics.GetSnapshot().Faults);
Assert.Equal(1, workerClient.KillCount);
Assert.Equal(1, workerClient.DisposeCount);
Assert.NotEqual(SessionState.Ready, starting.State);
@@ -1188,11 +1195,13 @@ public sealed class SessionManagerTests
/// <summary>
/// Same window as the kill case, driven by the host drain: a session still starting its
/// worker when <c>ShutdownAsync</c> runs is closed by the drain, and the worker client that
/// arrives afterwards is discarded rather than left running as an orphan.
/// arrives afterwards is discarded rather than left running as an orphan. The refused open
/// reports <c>GatewayStopping</c> (gRPC <c>Unavailable</c>, retry against the restarted
/// gateway), like an open refused at the door, and is not recorded as a fault.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task OpenSessionAsync_WhenShutdownDuringWorkerStartup_FailsOpenAndDiscardsLateWorkerClient()
public async Task OpenSessionAsync_WhenShutdownDuringWorkerStartup_ThrowsGatewayStoppingAndDiscardsLateWorkerClient()
{
FakeWorkerClient workerClient = new();
GatedSessionWorkerClientFactory factory = new(workerClient);
@@ -1207,7 +1216,8 @@ public sealed class SessionManagerTests
factory.ReleaseCreate();
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () => await openTask);
Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
Assert.Equal(SessionManagerErrorCode.GatewayStopping, exception.ErrorCode);
Assert.Equal(0, metrics.GetSnapshot().Faults);
Assert.Equal(SessionState.Closed, starting.State);
Assert.Equal(1, workerClient.KillCount);
Assert.Equal(1, workerClient.DisposeCount);