diff --git a/docs/Sessions.md b/docs/Sessions.md index 4ebb1ca..8502c96 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -207,6 +207,12 @@ The repair transitions the monitor's reconcile broadcasts on the alarm feed (Rai Sessions open with `MxGateway:Sessions:DefaultLeaseSeconds` (default 1800) added to the open timestamp. Unary client activity refreshes the lease by the same duration. `ExtendLease` and `IsLeaseExpired` cooperate with `SessionManager.CloseExpiredLeasesAsync`, which iterates a registry snapshot and closes any session whose lease has expired with `LeaseExpiredReason`. `SessionLeaseMonitorHostedService` runs that sweep every `MxGateway:Sessions:LeaseSweepIntervalSeconds` seconds (default 30). +#### Teardown parallelism + +A sweep pass is two phases. *Selection* stays a single sequential pass over the snapshot, because that is what gives the precedence rule (lease-expiry, then faulted, then detach-grace) and the `TryBeginCloseIfExpired` TOCTOU re-check their meaning. *Closing* then runs over the already-selected set with `Parallel.ForEachAsync` at `MaxParallelSessionCloses` (4). Each close is bounded by `MxGateway:Worker:ShutdownTimeoutSeconds` (default 10), so a one-at-a-time sweep lets a few hung workers serialize reaping and starve session slots for the rest. Parallel closing is safe because `TryBeginCloseIfExpired` already flipped each selected session to `Closing` under its own lock — that idempotent begin-close is the per-session exclusivity invariant, so no two teardowns can ever run against one session. The degree is bounded rather than unlimited because every concurrent close is one x86 worker process being shut down or killed; the fan-out exists to hide a few hung workers, not to tear the whole registry down at once. + +A close that throws no longer abandons the rest of the selected set: the sweep captures the first failure, finishes the remaining closes, then rethrows it so `SessionLeaseMonitorHostedService` still logs the sweep failure as before. `ShutdownAsync` drains sessions with the same bounded fan-out, keeping its existing per-session catch → `KillWorkerAsync` fallback. + #### Detach-grace retention `MxGateway:Sessions:DetachGraceSeconds` (default 30) is a bounded retention window kept after a session's *last external (gRPC) event-stream subscriber* drops, so a client can reconnect to the same session instead of having it torn down on the first stream disconnect. While the window is open the session stays `Ready` and fully usable — worker commands continue to work and a reconnecting subscriber re-attaches normally. Because retention is keyed on the *external* subscriber count (`_activeEventSubscriberCount`), and the gateway-owned internal dashboard mirror registers directly on the distributor with `isInternal: true` and is therefore *not* counted, a session whose only remaining subscriber is the dashboard mirror still enters detach-grace. @@ -276,16 +282,21 @@ If both graceful shutdown and the kill fall-back fail, the original and kill exc ## Shutdown Coordination -`SessionShutdownHostedService.StopAsync` calls `SessionManager.ShutdownAsync`, which closes every registered session with `GatewayShutdownReason`. The shutdown loop catches per-session exceptions, calls `KillWorker`, and removes the session so that one stuck worker cannot block the rest of the host: +`SessionShutdownHostedService.StopAsync` calls `SessionManager.ShutdownAsync`, which closes every registered session with `GatewayShutdownReason`. Sessions are drained with the same bounded fan-out the lease sweep uses (`MaxParallelSessionCloses`), because a one-at-a-time drain of a full registry at a worst-case worker shutdown timeout each outruns any host stop-timeout and leaves the tail to the orphan killer. Each iteration catches its own exceptions, calls `KillWorkerAsync`, and removes the session so that one stuck worker cannot block the rest of the host: ```csharp -public async Task ShutdownAsync(CancellationToken cancellationToken) -{ - foreach (GatewaySession session in _registry.Snapshot()) +await Parallel.ForEachAsync( + _registry.Snapshot(), + new ParallelOptions + { + MaxDegreeOfParallelism = MaxParallelSessionCloses, + CancellationToken = cancellationToken, + }, + async (session, closeToken) => { try { - await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false); + await CloseSessionCoreAsync(session, GatewayShutdownReason, closeToken).ConfigureAwait(false); } catch (Exception exception) { @@ -295,15 +306,13 @@ public async Task ShutdownAsync(CancellationToken cancellationToken) session.SessionId); if (_registry.TryGet(session.SessionId, out _)) { - session.KillWorker(GatewayShutdownReason); - await RemoveSessionAsync(session).ConfigureAwait(false); + await KillWorkerAsync(session.SessionId, GatewayShutdownReason, closeToken).ConfigureAwait(false); } } - } -} + }).ConfigureAwait(false); ``` -Iterating over `Snapshot` rather than the live dictionary lets `RemoveSessionAsync` mutate the registry inside the loop without throwing. +Iterating over `Snapshot` rather than the live dictionary lets `RemoveSessionAsync` mutate the registry from inside the loop without throwing, and gives the parallel drain a stable, already-materialized source. ## Dependency Injection diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs index f5c9a6d..ecff690 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs @@ -1,4 +1,5 @@ using System.Diagnostics.CodeAnalysis; +using System.Runtime.ExceptionServices; using System.Security.Cryptography; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Logging; @@ -20,6 +21,12 @@ public sealed class SessionManager : ISessionManager public const string DetachGraceExpiredReason = "detach-grace-expired"; public const string FaultedReason = "faulted-reaped"; + // Bounded so a mass expiry (or a host stop with a full registry) cannot stampede + // worker-process teardown: every concurrent close is one x86 worker being shut down or + // killed, and the point of the fan-out is to hide a few hung workers, not to tear the whole + // registry down at once. + private const int MaxParallelSessionCloses = 4; + private readonly ISessionRegistry _registry; private readonly ISessionWorkerClientFactory _workerClientFactory; private readonly GatewayMetrics _metrics; @@ -252,7 +259,10 @@ public sealed class SessionManager : ISessionManager DateTimeOffset now, CancellationToken cancellationToken) { - int closedCount = 0; + // Selection phase — deliberately sequential. Only the close calls below run in parallel: + // deciding WHICH sessions to close must stay a single ordered pass so the sweep-precedence + // rule and the TOCTOU re-check keep their meaning. + List<(GatewaySession Session, string Reason)> selected = []; foreach (GatewaySession session in _registry.Snapshot()) { // A session is swept when its normal lease has expired, it has FAULTED (a faulted @@ -288,45 +298,100 @@ public sealed class SessionManager : ISessionManager continue; } - await CloseSessionCoreAsync(session, reason, cancellationToken).ConfigureAwait(false); - closedCount++; + selected.Add((session, reason)); } + if (selected.Count == 0) + { + return 0; + } + + int closedCount = 0; + object failureSyncRoot = new(); + ExceptionDispatchInfo? firstFailure = null; + + // Close phase. Each close is bounded by the worker shutdown timeout (default 10 s), so a + // mass expiry with a few hung workers would serialize reaping and starve session slots. + // Parallel close is safe because TryBeginCloseIfExpired above already flipped every + // selected session to Closing under its own lock — that idempotent begin-close is the + // per-session exclusivity invariant, so no two teardowns can run against one session and + // a session selected here cannot be re-selected by a concurrent sweep. + await Parallel.ForEachAsync( + selected, + new ParallelOptions + { + MaxDegreeOfParallelism = MaxParallelSessionCloses, + CancellationToken = cancellationToken, + }, + async (candidate, closeToken) => + { + try + { + await CloseSessionCoreAsync(candidate.Session, candidate.Reason, closeToken).ConfigureAwait(false); + Interlocked.Increment(ref closedCount); + } + catch (Exception exception) + { + // The sequential sweep let a close failure propagate to the lease monitor, + // which logs it; that signal is preserved by rethrowing the first failure + // below. It is captured rather than thrown here so one failed (or hung) + // teardown does not abandon the rest of the already-selected set. + lock (failureSyncRoot) + { + firstFailure ??= ExceptionDispatchInfo.Capture(exception); + } + } + }).ConfigureAwait(false); + + firstFailure?.Throw(); + return closedCount; } /// public async Task ShutdownAsync(CancellationToken cancellationToken) { - foreach (GatewaySession session in _registry.Snapshot()) - { - try + // Sessions are drained in parallel: at a worst-case worker shutdown timeout each, a + // one-at-a-time drain of a full registry outruns any host stop-timeout and leaves the + // tail to the orphan killer. Per-session exclusivity comes from GatewaySession.CloseAsync's + // own close gate, and each iteration touches only its own session plus thread-safe + // registry/metrics state. + await Parallel.ForEachAsync( + _registry.Snapshot(), + new ParallelOptions { - await CloseSessionCoreAsync(session, GatewayShutdownReason, cancellationToken).ConfigureAwait(false); - } - catch (Exception exception) + MaxDegreeOfParallelism = MaxParallelSessionCloses, + CancellationToken = cancellationToken, + }, + async (session, closeToken) => { - _logger.LogWarning( - exception, - "Graceful shutdown failed for session {SessionId}; killing worker.", - session.SessionId); - - if (_registry.TryGet(session.SessionId, out _)) + try { - try + await CloseSessionCoreAsync(session, GatewayShutdownReason, closeToken).ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.LogWarning( + exception, + "Graceful shutdown failed for session {SessionId}; killing worker.", + session.SessionId); + + if (_registry.TryGet(session.SessionId, out _)) { - await KillWorkerAsync(session.SessionId, GatewayShutdownReason, cancellationToken).ConfigureAwait(false); - } - catch (SessionManagerException killException) - { - _logger.LogWarning( - killException, - "Worker kill fallback failed for session {SessionId}.", - session.SessionId); + try + { + await KillWorkerAsync(session.SessionId, GatewayShutdownReason, closeToken).ConfigureAwait(false); + } + catch (SessionManagerException killException) + { + _logger.LogWarning( + killException, + "Worker kill fallback failed for session {SessionId}.", + session.SessionId); + } } } - } - } + }).ConfigureAwait(false); } private async Task CloseSessionCoreAsync( diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs index c320ff1..f978ae1 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs @@ -1087,6 +1087,90 @@ public sealed class SessionManagerTests Assert.Equal(1, workerClient.ShutdownCount); } + /// + /// A sweep pass tears the selected sessions down concurrently rather than one worker + /// shutdown after another: with a mass expiry, a few hung workers would otherwise + /// serialize reaping (each close is bounded by Worker:ShutdownTimeoutSeconds) and + /// starve session slots. Overlap is asserted by counting concurrent entries into the fake + /// worker's shutdown rather than by wall clock, which is sturdier on a loaded box. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CloseExpiredLeasesAsync_ClosesExpiredSessionsConcurrently() + { + ShutdownConcurrencyProbe probe = new(expectedConcurrency: 2); + FakeWorkerClient firstClient = new() { ShutdownConcurrencyProbe = probe }; + FakeWorkerClient secondClient = new() { ShutdownConcurrencyProbe = probe }; + SessionManager manager = CreateManager(new QueueingSessionWorkerClientFactory(firstClient, secondClient)); + GatewaySession firstSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + GatewaySession secondSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None); + DateTimeOffset now = DateTimeOffset.UtcNow; + firstSession.ExtendLease(now.AddSeconds(-1)); + secondSession.ExtendLease(now.AddSeconds(-1)); + + int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None); + + Assert.Equal(2, closedCount); + Assert.Equal(SessionState.Closed, firstSession.State); + Assert.Equal(SessionState.Closed, secondSession.State); + Assert.Equal(2, probe.MaxObservedConcurrency); + } + + /// + /// Host stop drains sessions concurrently: 50 sessions at a worst-case 10 s shutdown each + /// would exceed any host stop-timeout if drained one at a time, leaving the tail to the + /// orphan killer. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ShutdownAsync_ClosesSessionsConcurrently() + { + ShutdownConcurrencyProbe probe = new(expectedConcurrency: 2); + FakeWorkerClient firstClient = new() { ShutdownConcurrencyProbe = probe }; + FakeWorkerClient secondClient = new() { ShutdownConcurrencyProbe = probe }; + SessionManager manager = CreateManager(new QueueingSessionWorkerClientFactory(firstClient, secondClient)); + GatewaySession firstSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + GatewaySession secondSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None); + + await manager.ShutdownAsync(CancellationToken.None); + + Assert.Equal(SessionState.Closed, firstSession.State); + Assert.Equal(SessionState.Closed, secondSession.State); + Assert.Equal(2, probe.MaxObservedConcurrency); + } + + /// + /// A close that throws must not abandon the rest of the selected set: the sweep still + /// tears the healthy expired session down, and the failure still surfaces to the lease + /// monitor (which logs it) exactly as the sequential loop did. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CloseExpiredLeasesAsync_WhenOneCloseFails_StillClosesRemainingSessionsAndRethrows() + { + FakeWorkerClient failingClient = new() + { + ShutdownException = new InvalidOperationException("worker shutdown failed"), + KillException = new InvalidOperationException("worker kill failed"), + }; + FakeWorkerClient healthyClient = new(); + SessionManager manager = CreateManager(new QueueingSessionWorkerClientFactory(failingClient, healthyClient)); + GatewaySession failingSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + GatewaySession healthySession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None); + DateTimeOffset now = DateTimeOffset.UtcNow; + failingSession.ExtendLease(now.AddSeconds(-1)); + healthySession.ExtendLease(now.AddSeconds(-1)); + + SessionManagerException exception = await Assert.ThrowsAsync( + async () => await manager.CloseExpiredLeasesAsync(now, CancellationToken.None)); + + Assert.Equal(SessionManagerErrorCode.CloseFailed, exception.ErrorCode); + Assert.Equal(1, healthyClient.ShutdownCount); + Assert.Equal(SessionState.Closed, healthySession.State); + Assert.False(manager.TryGetSession(healthySession.SessionId, out _)); + Assert.False(manager.TryGetSession(failingSession.SessionId, out _)); + } + /// Verifies that shutdown closes all registered sessions. /// A task that represents the asynchronous operation. [Fact] @@ -1274,6 +1358,13 @@ public sealed class SessionManagerTests /// Gets a value indicating whether to block shutdown on the fake worker client. public bool BlockShutdown { get; init; } + /// + /// Gets the rendezvous that records how many shutdowns overlap, shared by the fakes of + /// the sessions a single teardown pass closes. Null when the test does not measure + /// teardown concurrency. + /// + public ShutdownConcurrencyProbe? ShutdownConcurrencyProbe { get; init; } + /// Gets the last command invoked on the fake worker client. public WorkerCommand? LastCommand { get; private set; } @@ -1335,6 +1426,11 @@ public sealed class SessionManagerTests throw ShutdownException; } + if (ShutdownConcurrencyProbe is not null) + { + await ShutdownConcurrencyProbe.EnterAsync(cancellationToken); + } + if (BlockShutdown) { ShutdownStarted.TrySetResult(); @@ -1379,4 +1475,65 @@ public sealed class SessionManagerTests } } + /// + /// Rendezvous that measures how many worker shutdowns a teardown pass runs at once. Each + /// entering shutdown records the in-flight count and waits until + /// shutdowns are in flight, so a genuinely parallel teardown releases immediately while a + /// sequential one can only release on the bounded timeout — with a max observed concurrency + /// of one, which is the assertion that fails. + /// + /// Number of overlapping shutdowns that releases the rendezvous. + private sealed class ShutdownConcurrencyProbe(int expectedConcurrency) + { + private static readonly TimeSpan RendezvousTimeout = TimeSpan.FromSeconds(5); + + private readonly TaskCompletionSource _reached = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _inFlight; + private int _maxInFlight; + + /// Gets the highest number of shutdowns observed in flight at the same time. + public int MaxObservedConcurrency => Volatile.Read(ref _maxInFlight); + + /// Enters the rendezvous for one worker shutdown and waits for the expected overlap. + /// Token that abandons the wait. + /// A task that represents the asynchronous operation. + public async Task EnterAsync(CancellationToken cancellationToken) + { + int inFlight = Interlocked.Increment(ref _inFlight); + RecordMax(inFlight); + if (inFlight >= expectedConcurrency) + { + _reached.TrySetResult(); + } + + try + { + await _reached.Task.WaitAsync(RendezvousTimeout, cancellationToken); + } + catch (TimeoutException) + { + // Sequential teardown: the expected overlap never happens, so let the shutdown + // finish and let MaxObservedConcurrency report the (failing) truth. + } + finally + { + Interlocked.Decrement(ref _inFlight); + } + } + + private void RecordMax(int inFlight) + { + int observed = Volatile.Read(ref _maxInFlight); + while (inFlight > observed) + { + int previous = Interlocked.CompareExchange(ref _maxInFlight, inFlight, observed); + if (previous == observed) + { + return; + } + + observed = previous; + } + } + } }