dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(sessions): refuse OpenSession while the gateway is stopping; drain until the registry is empty

Browse Source 
ShutdownAsync iterated a single registry snapshot while Kestrel still
accepted OpenSession calls (ApplicationStopping fires before hosted
services stop), so a session opened behind the snapshot had nobody left to
close it and its x86 worker outlived the gateway as an orphan.

SessionManager now observes IHostApplicationLifetime.ApplicationStopping
(and sets the same flag at the start of ShutdownAsync) and rejects
OpenSessionAsync with the new SessionManagerErrorCode.GatewayStopping,
re-checked after registration so the flag write before the drain's
snapshot closes the race. GatewayStopping maps to gRPC Unavailable and is
not a fault. ShutdownAsync re-snapshots until the registry is empty,
bounded at three passes.
This commit is contained in:
Joseph Doherty
2026-10-22 06:23:40 -04:00
parent 09666cc719
commit 37ab629d70
6 changed files with 317 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs
+114 -38
View File
@@ -2,6 +2,7 @@ using System.Diagnostics.CodeAnalysis;
using System.Runtime.ExceptionServices;
using System.Security.Cryptography;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
@@ -27,6 +28,13 @@ public sealed class SessionManager : ISessionManager
// registry down at once.
private const int MaxParallelSessionCloses = 4;
// ShutdownAsync re-snapshots the registry until it is empty so an open that registered its
// session just after a snapshot is still drained. The stopping gate in OpenSessionAsync is
// what actually closes that window (it is re-checked after registration), so the loop is a
// backstop and a small bound is enough; anything still registered after the last pass is
// logged and left to the orphan terminator on the next start.
private const int MaxShutdownDrainPasses = 3;
private readonly ISessionRegistry _registry;
private readonly ISessionWorkerClientFactory _workerClientFactory;
private readonly GatewayMetrics _metrics;
@@ -39,6 +47,12 @@ public sealed class SessionManager : ISessionManager
private readonly Dashboard.Hubs.IDashboardEventBroadcaster? _dashboardEventBroadcaster;
private readonly ArrayAddressNormalizer? _addressNormalizer;
// 0 = accepting opens, 1 = stopping. Set by the host's ApplicationStopping signal (which fires
// before hosted services stop, while Kestrel still accepts gRPC calls) and, as a fallback for
// hosts without a lifetime, at the start of ShutdownAsync. Never reset: a manager that has
// begun draining does not come back.
private int _stopping;
/// <summary>
/// Initializes a new instance of <see cref="SessionManager"/>.
/// </summary>
@@ -61,6 +75,12 @@ public sealed class SessionManager : ISessionManager
/// metadata; handed to each session so the normalization runs at the outbound choke point.
/// Null in unit tests that do not exercise array-write ergonomics.
/// </param>
/// <param name="applicationLifetime">
/// Host lifetime whose <c>ApplicationStopping</c> closes the open gate as soon as the host
/// begins stopping — before hosted services stop and while Kestrel is still accepting gRPC
/// calls — so no session is opened behind the shutdown drain. Null in unit tests; the gate
/// then closes when <see cref="ShutdownAsync"/> starts.
/// </param>
public SessionManager(
ISessionRegistry registry,
ISessionWorkerClientFactory workerClientFactory,
@@ -71,7 +91,8 @@ public sealed class SessionManager : ISessionManager
Grpc.MxAccessGrpcMapper? eventMapper = null,
ILogger<SessionEventDistributor>? distributorLogger = null,
Dashboard.Hubs.IDashboardEventBroadcaster? dashboardEventBroadcaster = null,
ArrayAddressNormalizer? addressNormalizer = null)
ArrayAddressNormalizer? addressNormalizer = null,
IHostApplicationLifetime? applicationLifetime = null)
{
_registry = registry ?? throw new ArgumentNullException(nameof(registry));
_workerClientFactory = workerClientFactory ?? throw new ArgumentNullException(nameof(workerClientFactory));
@@ -85,6 +106,10 @@ public sealed class SessionManager : ISessionManager
_addressNormalizer = addressNormalizer;
_options = options.Value;
_sessionSlots = new SemaphoreSlim(_options.Sessions.MaxSessions, _options.Sessions.MaxSessions);
// The registration is never disposed: the manager is a host singleton that lives exactly
// as long as the lifetime it observes.
applicationLifetime?.ApplicationStopping.Register(MarkStopping);
}
/// <inheritdoc />
@@ -104,6 +129,7 @@ public sealed class SessionManager : ISessionManager
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(request);
ThrowIfStopping();
EnsureSessionCapacity();
GatewaySession? session = null;
@@ -119,6 +145,12 @@ public sealed class SessionManager : ISessionManager
}
registered = true;
// Re-check after registration. The stopping flag is written before ShutdownAsync takes
// its registry snapshot, so an open that does not observe it here registered its
// session before that snapshot and will be drained; one that does observe it rolls
// back itself. Either way no session is created behind the drain.
ThrowIfStopping();
session.TransitionTo(SessionState.StartingWorker);
IWorkerClient workerClient = await _workerClientFactory
.CreateAsync(session, cancellationToken)
@@ -140,6 +172,13 @@ public sealed class SessionManager : ISessionManager
return session;
}
catch (SessionManagerException exception) when (exception.ErrorCode == SessionManagerErrorCode.GatewayStopping)
{
// Not an open failure: the gateway is going away and the caller should retry against
// the restarted one, so no fault is recorded and the code is surfaced as-is.
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
throw;
}
catch (Exception exception)
{
await RollBackOpenAsync(session, registered, exception).ConfigureAwait(false);
@@ -435,49 +474,86 @@ public sealed class SessionManager : ISessionManager
// intentional: that sweep is periodic maintenance whose missed sessions are picked up by
// the next pass and, ultimately, by this drain. This drain is terminal — nothing runs after
// it — so it must not be abandoned partway.
await Parallel.ForEachAsync(
_registry.Snapshot(),
new ParallelOptions { MaxDegreeOfParallelism = MaxParallelSessionCloses },
async (session, _) =>
//
// Close the open gate BEFORE the first snapshot (see the re-check in OpenSessionAsync), then
// re-snapshot until the registry is empty so an open that registered just before the gate
// closed is drained too.
MarkStopping();
for (int pass = 0; pass < MaxShutdownDrainPasses; pass++)
{
if (_registry.Count == 0)
{
try
{
await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.LogWarning(
exception,
"Graceful shutdown failed for session {SessionId}; killing worker.",
session.SessionId);
return;
}
// Defensive fallback: CloseSessionCoreAsync's inner
// SessionCloseStartedException catch normally removes the session
// and accounts the close. This outer fallback only fires for
// sessions still in the registry — route through KillWorkerAsync
// so the bookkeeping is identical to the dashboard kill path.
if (_registry.TryGet(session.SessionId, out GatewaySession? registeredSession)
&& registeredSession is not null)
await Parallel.ForEachAsync(
_registry.Snapshot(),
new ParallelOptions { MaxDegreeOfParallelism = MaxParallelSessionCloses },
async (session, _) =>
{
try
{
try
await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.LogWarning(
exception,
"Graceful shutdown failed for session {SessionId}; killing worker.",
session.SessionId);
// Defensive fallback: CloseSessionCoreAsync's inner
// SessionCloseStartedException catch normally removes the session
// and accounts the close. This outer fallback only fires for
// sessions still in the registry — route through KillWorkerAsync
// so the bookkeeping is identical to the dashboard kill path.
if (_registry.TryGet(session.SessionId, out GatewaySession? registeredSession)
&& registeredSession is not null)
{
// Deliberately NOT the caller's token: the kill is the last-resort orphan
// preventer, so it must still run when the host stop deadline (or a
// sibling body's failure) has already cancelled the drain. It is a
// synchronous Kill plus registry/dispose bookkeeping, not a wait on
// the worker, so it cannot extend the drain meaningfully.
await KillWorkerAsync(session.SessionId, GatewayShutdownReason, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception killException)
{
_logger.LogWarning(
killException,
"Worker kill fallback failed for session {SessionId}.",
session.SessionId);
try
{
// Deliberately NOT the caller's token: the kill is the last-resort orphan
// preventer, so it must still run when the host stop deadline (or a
// sibling body's failure) has already cancelled the drain. It is a
// synchronous Kill plus registry/dispose bookkeeping, not a wait on
// the worker, so it cannot extend the drain meaningfully.
await KillWorkerAsync(session.SessionId, GatewayShutdownReason, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception killException)
{
_logger.LogWarning(
killException,
"Worker kill fallback failed for session {SessionId}.",
session.SessionId);
}
}
}
}
}).ConfigureAwait(false);
}).ConfigureAwait(false);
}
int remaining = _registry.Count;
if (remaining > 0)
{
_logger.LogWarning(
"Session drain finished with {Remaining} session(s) still registered after {Passes} passes; their workers are left to the orphan terminator on the next start.",
remaining,
MaxShutdownDrainPasses);
}
}
private void MarkStopping()
{
Volatile.Write(ref _stopping, 1);
}
private void ThrowIfStopping()
{
if (Volatile.Read(ref _stopping) != 0)
{
throw new SessionManagerException(
SessionManagerErrorCode.GatewayStopping,
"The gateway is stopping and is not accepting new sessions.");
}
}
private async Task<SessionCloseResult> CloseSessionCoreAsync(