diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index ba4dc02..96b16ca 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -154,7 +154,7 @@ validation fails fast when either half of the headroom rule is violated. | `MxGateway:Sessions:LeaseSweepIntervalSeconds` | `30` | Hosted monitor interval for closing expired leases. Active event-stream subscribers keep a session from expiring while the stream remains attached. | | `MxGateway:Sessions:DetachGraceSeconds` | `30` | Detach-grace retention window. When positive, a session whose last external (gRPC) event-stream subscriber drops is retained in `Ready` for this many seconds so a client can reconnect; if no external subscriber re-attaches within the window, the lease monitor closes it with `detach-grace-expired`. The internal dashboard mirror does not count as an external subscriber, so a dashboard-only session still enters detach-grace. `0` disables retention and reverts to closing only on normal lease expiry. Must be zero or greater. Reconnect/replay itself is implemented separately (Task 12); this option controls retention and expiry only. The effective close happens within the next sweep cycle after the window elapses — up to `LeaseSweepIntervalSeconds` after expiry. Operators wanting a firm minimum retention bound should set `DetachGraceSeconds` greater than `LeaseSweepIntervalSeconds`. | | `MxGateway:Sessions:FaultedGraceSeconds` | `0` | Grace window before the lease monitor reaps a faulted session (killing its worker and freeing the slot). A faulted session is permanently unusable yet otherwise pins a session slot and a live x86 worker until its normal lease expires. `0` (the default) reaps it on the next sweep cycle, bounding blast radius; a positive value keeps the faulted session observable via `GetSessionStatus` for that window before it is reclaimed, with the close reason `faulted-reaped`. Must be zero or greater. | -| `MxGateway:Sessions:UnattachedSessionGraceSeconds` | `0` | Idle window after which the lease monitor reaps a `Ready` session that has **never** attached a `StreamEvents` subscriber, with the close reason `unattached-grace-expired`. Exists because a client that opens a session and vanishes before it ever streams — OtOpcUa's `GalaxyDriver` failing part-way through start-up, for example — otherwise pins a session slot and a live x86 worker for the full `DefaultLeaseSeconds` (30 minutes). The window is measured from the session's last client activity (open, or its most recent unary command), so a unary-only client that keeps calling is never reaped by this rule, and it stops applying once the session's first subscriber attaches (`DetachGraceSeconds` and the normal lease govern it from then on). `0` (the default) disables it. Must be zero or greater. Set it comfortably above the time a well-behaved client needs between `OpenSession` and its first `StreamEvents`; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. | +| `MxGateway:Sessions:UnattachedSessionGraceSeconds` | `0` | Idle window after which the lease monitor reaps a `Ready` session that has **never** attached a `StreamEvents` subscriber, with the close reason `unattached-grace-expired`. Exists because a client that opens a session and vanishes before it ever streams — OtOpcUa's `GalaxyDriver` failing part-way through start-up, for example — otherwise pins a session slot and a live x86 worker for the full `DefaultLeaseSeconds` (30 minutes). The window is measured from the later of the session becoming `Ready` and its last unary command, so a slow worker start does not eat it and a unary-only client that keeps calling is never reaped by this rule; it stops applying once the session's first subscriber attaches (`DetachGraceSeconds` and the normal lease govern it from then on), and the gateway's own alarm-monitor session, whose subscriber is internal, is exempt the same way. `0` (the default) disables it. Must be zero or greater. Set it comfortably above the time a well-behaved client needs between `OpenSession` and its first `StreamEvents`; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. | | `MxGateway:Sessions:AllowMultipleEventSubscribers` | `false` | Controls whether multiple `StreamEvents` subscribers may attach to one session. When `false` the session refuses a second subscriber with `AlreadyExists`. Set to `true` to enable fan-out via the `SessionEventDistributor`. | | `MxGateway:Sessions:MaxEventSubscribersPerSession` | `8` | Maximum number of concurrent `StreamEvents` subscribers per session when `AllowMultipleEventSubscribers` is `true`. Effectively 1 when `AllowMultipleEventSubscribers` is `false`. Must be greater than zero. | | `MxGateway:Sessions:WorkerReadyWaitTimeoutMs` | `0` | Bounded time, in milliseconds, the gateway will wait for a worker to reach `Ready` when the session is already `Ready` but the worker state has transiently diverged (e.g. `Handshaking` after a heartbeat blip). Applies only to transient worker states; terminal states (`Faulted`/`Closing`/`Closed`/no worker) fail fast immediately regardless of this setting. `0` (the default) disables the wait and preserves the original fail-fast behavior. Must be greater than or equal to zero. | diff --git a/docs/Sessions.md b/docs/Sessions.md index 9ca4742..b945fd0 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -199,6 +199,8 @@ The session is registered — and therefore visible to the dashboard's Kill butt `GatewaySession.AttachWorkerClient` therefore returns `false`, under `_syncRoot`, when the session is already `Closing`, `Closed` or `Faulted` or has been disposed. The manager runs the end of the open sequence — attach, `MarkReady`, `SessionOpened` — through `TryAttachWorkerClientAndMarkReadyAsync`, which holds the per-session close gate (`_closeLock`) so a concurrent `CloseAsync` / `KillWorkerWithCloseGateAsync` runs either entirely before it (the attach is refused) or entirely after it (an ordinary close of a Ready session). On a refused attach the manager kills and disposes the late worker client and the open fails with `OpenFailed`; the registry is already empty and the slot was released by whichever teardown closed the session. +Once `CreateAsync` has returned, a live worker exists that only the open path knows about, so the gate wait is **not cancellable** (it waits on `CancellationToken.None`; the gate is only ever held for the duration of one close or kill) and the manager discards the client on *any* exception out of the attach step — the caller's own cancellation (checked before the attach, so an abandoned open never yields a session), a disposed gate, anything thrown while becoming Ready. The failure then surfaces as `OpenFailed` with the cause as inner exception. An exception that escaped before the discard would be an orphaned x86 worker, which a restarted gateway terminates rather than adopts. + Because such a session was never counted as opened, it must not be counted as closed either. `GatewaySession.OpenRecorded` is set inside the same gate right after `SessionOpened` is recorded, and every teardown path (`KillWorkerAsync`, `CloseSessionCoreAsync`, the close-failed branch) records `SessionClosed` / `SessionRemoved` only when it is set — so `sessions.opened`, `sessions.closed` and the `sessions.open` gauge stay consistent across the startup window. ### Run @@ -251,9 +253,9 @@ A session that faults (for example, a slow single-subscriber client that overflo #### Never-subscribed session reaping -Detach-grace only helps a session that *had* a subscriber. A client that opens a session and then disappears before it ever calls `StreamEvents` — OtOpcUa's `GalaxyDriver` failing part-way through its start-up is the motivating case — leaves a `Ready` session with no subscriber, no detach timestamp and a lease 30 minutes out, pinning one of `MaxSessions` and a live x86 worker for all of `DefaultLeaseSeconds`. `MxGateway:Sessions:UnattachedSessionGraceSeconds` (default `0`, off) closes that gap: when positive, `CloseExpiredLeasesAsync` also checks `IsUnattachedGraceExpired(now)` and reaps, with the distinct `UnattachedGraceExpiredReason` (`"unattached-grace-expired"`), any `Ready` session that has never had an external subscriber (the same `_everHadEventSubscriber` latch that gates detach-grace) and has been idle for the window. +Detach-grace only helps a session that *had* a subscriber. A client that opens a session and then disappears before it ever calls `StreamEvents` — OtOpcUa's `GalaxyDriver` failing part-way through its start-up is the motivating case — leaves a `Ready` session with no subscriber, no detach timestamp and a lease 30 minutes out, pinning one of `MaxSessions` and a live x86 worker for all of `DefaultLeaseSeconds`. `MxGateway:Sessions:UnattachedSessionGraceSeconds` (default `0`, off) closes that gap: when positive, `CloseExpiredLeasesAsync` also checks `IsUnattachedGraceExpired(now)` and reaps, with the distinct `UnattachedGraceExpiredReason` (`"unattached-grace-expired"`), any `Ready` session that has never had a subscriber of either kind and has been idle for the window. -Three deliberate limits keep the rule from catching legitimate clients. It applies only to `Ready` sessions, so a slow worker start is never reaped mid-startup. Idle is measured from the session's *last client activity* — open, or its most recent unary command — not from open, so a unary-only client (reads and writes, never streams) is not reaped while it keeps calling; the rule is an idle timeout for sessions that never streamed, not a deadline to stream. And it stops applying the moment the first subscriber attaches, after which detach-grace and the normal lease govern the session. Like the other sweep conditions it is re-verified inside `TryBeginCloseIfExpired`, so a subscriber that attaches between selection and the close wins the race. Operators should set the window comfortably above a well-behaved client's open-to-first-`StreamEvents` time; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. +Four deliberate limits keep the rule from catching legitimate sessions. It applies only to `Ready` sessions, and its idle clock starts at `Ready` (`MarkReady` stamps `_readyAtUtc`), so a slow worker start neither reaps the session mid-startup nor eats the window before the client even has the session id. Idle is measured from the later of that Ready stamp and the session's *last client activity* (its most recent unary command), so a unary-only client (reads and writes, never streams) is not reaped while it keeps calling; the rule is an idle timeout for sessions that never streamed, not a deadline to stream. It stops applying the moment the first subscriber attaches, after which detach-grace and the normal lease govern the session. And an *internal* in-process subscriber counts as attached for this rule (`AttachInternalEventSubscriber` sets a separate `_everHadInternalEventSubscriber` latch): the gateway's own `GatewayAlarmMonitor` session only ever attaches that way, and it must not be reaped by its own gateway every cycle. The latch is separate from `_everHadEventSubscriber` because that one drives detach-grace, which is keyed on external subscribers only. The dashboard mirror, which every session registers at `MarkReady`, is deliberately *not* a subscriber for this rule. Like the other sweep conditions it is re-verified inside `TryBeginCloseIfExpired`, so a subscriber that attaches between selection and the close wins the race. Operators should set the window comfortably above a well-behaved client's open-to-first-`StreamEvents` time; the close lands on the next sweep after the window elapses, up to `LeaseSweepIntervalSeconds` later. #### Reconnect and replay @@ -314,6 +316,8 @@ A kill that succeeds completes the close **successfully**: the worker is gone an Only if both the graceful shutdown and the kill fall-back fail are the original and kill exceptions bundled into an `AggregateException` and surfaced as `SessionCloseStartedException`. `SessionManager.CloseSessionCoreAsync` then translates that into a `SessionManagerException` with `CloseFailed`, faults the session, records the fault metric and removes the session — a worker may still be running, which is what `CloseFailed` is reserved for. +**Losing a race to a concurrent close.** A caller that looked a session up (a kill, a client close, a sweep's snapshot) and then found its close gate already disposed — because a concurrent close finished, removed the session, accounted `SessionClosed` and disposed it in between — sees `ObjectDisposedException` from the gate. Both `KillWorkerAsync` and `CloseSessionCoreAsync` catch that case *when the session is no longer registered* and return `AlreadyClosed: true` with no accounting, no fault and no log: the close that won already did all of it. (The old kill catch recorded a fault, faulted the session and decremented the open gauge a second time; the sweep rethrew it as a sweep failure.) A sweep does not count such a session among the sessions it closed. A disposed gate on a session that *is* still registered is a real failure and takes the normal `CloseFailed` path. + `SessionManager.KillWorkerAsync` is the explicit kill path that the dashboard's admin Kill button invokes and that `ShutdownAsync` falls back to when a graceful close throws. It goes through `GatewaySession.KillWorkerWithCloseGateAsync`, which skips `WorkerClient.ShutdownAsync` entirely, so `KillCount` increments while `ShutdownCount` does not; the session is then removed from the registry and the open-session slot is released, identical to the cleanup that follows a successful `CloseSessionAsync`. ## Shutdown Coordination diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs index c4f4edf..0c311d1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs @@ -63,11 +63,13 @@ public sealed class SessionOptions /// and vanishes before it ever streams (for example OtOpcUa's GalaxyDriver failing /// part-way through its start-up) otherwise pins a session slot and a live x86 worker for /// the full (30 minutes by default). The window is - /// measured from the session's last client activity — open or the most recent unary - /// command — so a client that only ever reads and writes is not reaped while it keeps - /// calling, and it stops applying the moment the session's first subscriber attaches - /// (from then on and the normal lease govern it). A value - /// of 0 (the default) disables the sweep. Must be greater than or equal to zero. + /// measured from the later of the session becoming Ready and its most recent unary + /// command, so a slow worker start does not eat it and a client that only ever reads and + /// writes is not reaped while it keeps calling; it stops applying the moment the session's + /// first subscriber — external, or the gateway's own internal alarm-monitor subscriber — + /// attaches (from then on and the normal lease govern + /// it). A value of 0 (the default) disables the sweep. Must be greater than or + /// equal to zero. /// /// /// Set it comfortably above the time a well-behaved client needs between diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index f3a2525..8cdb36f 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -39,6 +39,12 @@ public sealed class GatewaySession // FAILED first attach — which still runs the rollback DetachEventSubscriber from the // attach catch path — does not push a never-subscribed session into the grace window. private bool _everHadEventSubscriber; + // True once an in-process (internal, non-counted) subscriber attached via + // AttachInternalEventSubscriber. Consulted only by the unattached-grace sweep rule. + private bool _everHadInternalEventSubscriber; + // Stamped once, when the session first becomes Ready; the unattached-grace idle clock cannot + // start before it. + private DateTimeOffset? _readyAtUtc; private SessionEventDistributor? _eventDistributor; private bool _eventDistributorStarted; // Set under _syncRoot when DisposeAsync nulls _eventDistributor, making the @@ -468,23 +474,27 @@ public sealed class GatewaySession /// Invoked inside the gate once the session is Ready; the caller records its open-session /// accounting here, and is set right after it returns. /// - /// Token to cancel waiting for the gate. /// /// when the session became Ready; when the /// attach was refused (see ), including when the session was /// disposed before the gate could be taken. /// + /// + /// 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. + /// internal async ValueTask TryAttachWorkerClientAndMarkReadyAsync( IWorkerClient workerClient, - Action onReady, - CancellationToken cancellationToken) + Action onReady) { ArgumentNullException.ThrowIfNull(workerClient); ArgumentNullException.ThrowIfNull(onReady); try { - await _closeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + await _closeLock.WaitAsync(CancellationToken.None).ConfigureAwait(false); } catch (ObjectDisposedException) { @@ -510,7 +520,15 @@ public sealed class GatewaySession } finally { - _closeLock.Release(); + try + { + _closeLock.Release(); + } + catch (ObjectDisposedException) + { + // A double-dispose that slipped past DisposeAsync's own gate; the session is torn + // down either way and there is no one left to hand the slot to. + } } } @@ -594,6 +612,14 @@ public sealed class GatewaySession public void MarkReady() { TransitionTo(SessionState.Ready); + lock (_syncRoot) + { + // The unattached-grace idle clock starts here, not at open: the client cannot + // subscribe before it has the session, and a slow worker start must not hand it a + // session the next sweep reaps. + _readyAtUtc ??= _eventStreaming.TimeProvider.GetUtcNow(); + } + StartDashboardMirror(); } @@ -747,6 +773,15 @@ public sealed class GatewaySession SessionEventDistributor distributor = EnsureDistributorCreated(out bool startNow); IEventSubscriberLease lease = distributor.Register(isInternal: true); StartPumpIfRequested(distributor, startNow); + lock (_syncRoot) + { + // An in-process consumer (the alarm monitor) is a real subscriber for the + // unattached-grace rule: its session is attached, not abandoned. It stays out of + // _everHadEventSubscriber on purpose — that latch drives detach-grace, which is keyed + // on EXTERNAL subscribers only. + _everHadInternalEventSubscriber = true; + } + return lease; } @@ -1903,14 +1938,26 @@ public sealed class GatewaySession // Ready only: a session still starting its worker is not idle, it is waiting, and reaping it // mid-startup would turn a slow worker start into an OpenFailed for a client that did nothing - // wrong. Idle is measured from the last client activity (open or last unary command) rather - // than from open so a unary-only client that keeps calling is never reaped by this rule. + // wrong. Idle is measured from the later of Ready and the last client activity (last unary + // command) rather than from open, so a slow start does not eat the window and a unary-only + // client that keeps calling is never reaped by this rule. An internal (in-process) subscriber + // — the gateway's own alarm monitor — counts as attached, or the gateway would reap its own + // monitor session every cycle. private bool IsUnattachedGraceExpiredCore(DateTimeOffset now) - => _unattachedGrace > TimeSpan.Zero - && _state is SessionState.Ready - && !_everHadEventSubscriber - && _activeEventSubscriberCount == 0 - && now - _lastClientActivityAt >= _unattachedGrace; + { + if (_unattachedGrace <= TimeSpan.Zero + || _state is not SessionState.Ready + || _readyAtUtc is null + || _everHadEventSubscriber + || _everHadInternalEventSubscriber + || _activeEventSubscriberCount != 0) + { + return false; + } + + DateTimeOffset idleSince = _lastClientActivityAt > _readyAtUtc.Value ? _lastClientActivityAt : _readyAtUtc.Value; + return now - idleSince >= _unattachedGrace; + } // Final terminal transition; under _syncRoot to keep _state writes single-lock. // Closed is unconditionally terminal — TransitionTo refuses to overwrite it — diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs index 625c4f1..7097ba6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs @@ -157,13 +157,27 @@ public sealed class SessionManager : ISessionManager .CreateAsync(session, cancellationToken) .ConfigureAwait(false); - // The session has been registered (and killable / drainable) for the whole worker - // startup. If a kill or the host drain closed it meanwhile, the attach is refused and - // the late worker is discarded here — it must not be returned on a dead session, and - // nothing else would ever stop it. - bool ready = await session - .TryAttachWorkerClientAndMarkReadyAsync(workerClient, _metrics.SessionOpened, cancellationToken) - .ConfigureAwait(false); + // From here on a live worker exists and nothing but this method knows about it: every + // way out of this block other than "attached and Ready" must discard (kill + dispose) + // it, or it is an orphan the gateway cannot reattach. The session has been registered + // (and killable / drainable) for the whole worker startup, so the attach is refused + // 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; + try + { + cancellationToken.ThrowIfCancellationRequested(); + ready = await session + .TryAttachWorkerClientAndMarkReadyAsync(workerClient, _metrics.SessionOpened) + .ConfigureAwait(false); + } + catch + { + await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false); + throw; + } + if (!ready) { await DiscardWorkerClientAsync(session, workerClient).ConfigureAwait(false); @@ -317,15 +331,23 @@ public sealed class SessionManager : ISessionManager { wasClosed = await session.KillWorkerWithCloseGateAsync(reason, cancellationToken).ConfigureAwait(false); } + catch (ObjectDisposedException) when (!_registry.TryGet(sessionId, out _)) + { + // Lost the race to a concurrent close: between the lookup above and the gate wait, the + // close finished, removed the session, accounted SessionClosed and disposed the gate. + // Nothing is left to kill and nothing to account; report it as already closed. + return new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: true); + } catch (Exception exception) { session.MarkFaulted(exception.Message); _metrics.Fault(SessionManagerErrorCode.CloseFailed.ToString()); // If the open-session gauge was incremented for this session (it was not when the - // kill landed while the worker was still starting), decrement it here so - // mxgateway.sessions.open does not leak when the kill path throws. - if (session.OpenRecorded) + // kill landed while the worker was still starting) and nobody has removed it yet + // (a removal accounts the gauge itself), decrement it here so mxgateway.sessions.open + // does not leak when the kill path throws. + if (session.OpenRecorded && _registry.TryGet(sessionId, out _)) { _metrics.SessionRemoved(); } @@ -428,8 +450,11 @@ public sealed class SessionManager : ISessionManager { try { - await CloseSessionCoreAsync(candidate.Session, candidate.Reason, closeToken).ConfigureAwait(false); - Interlocked.Increment(ref closedCount); + SessionCloseResult result = await CloseSessionCoreAsync(candidate.Session, candidate.Reason, closeToken).ConfigureAwait(false); + if (!result.AlreadyClosed) + { + Interlocked.Increment(ref closedCount); + } } catch (Exception exception) { @@ -594,6 +619,13 @@ public sealed class SessionManager : ISessionManager { throw; } + catch (ObjectDisposedException) when (!_registry.TryGet(session.SessionId, out _)) + { + // Lost the race to a concurrent close that removed, accounted and disposed the + // session after this caller picked it up (a sweep's snapshot, a client close that + // looked it up first). Not a failure and nothing to redo. + return new SessionCloseResult(session.SessionId, SessionState.Closed, AlreadyClosed: true); + } catch (SessionCloseStartedException exception) { session.MarkFaulted(exception.Message); 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 d50faf7..e82b83e 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs @@ -924,6 +924,71 @@ public sealed class SessionManagerTests Assert.True(metrics.GetSnapshot().Faults > 0); } + /// + /// A kill whose session lookup succeeded but whose close gate is disposed by the time it + /// waits on it lost the race to a close that already removed, disposed and accounted the + /// session. It must report AlreadyClosed and touch nothing: the old catch recorded a + /// fault, faulted the session and decremented the open gauge a second time (visible here as + /// the other open session's slot disappearing from the gauge). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task KillWorkerAsync_WhenCloseWonTheRace_ReportsAlreadyClosedWithoutDoubleAccounting() + { + FakeWorkerClient racedClient = new(); + FakeWorkerClient otherClient = new(); + StaleRegistry registry = new(); + using GatewayMetrics metrics = new(); + SessionManager manager = CreateManager(new QueueingSessionWorkerClientFactory(racedClient, otherClient), registry, metrics); + GatewaySession raced = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None); + + await manager.CloseSessionAsync(raced.SessionId, CancellationToken.None); + Assert.Equal(1, metrics.GetSnapshot().OpenSessions); + + // The kill looked the session up just before the close removed and disposed it. + registry.ServeStaleLookupOnce(raced); + SessionCloseResult result = await manager.KillWorkerAsync(raced.SessionId, "admin-kill", CancellationToken.None); + + Assert.True(result.AlreadyClosed); + Assert.Equal(SessionState.Closed, result.FinalState); + Assert.Equal(0, racedClient.KillCount); + GatewayMetricsSnapshot snapshot = metrics.GetSnapshot(); + Assert.Equal(1, snapshot.OpenSessions); + Assert.Equal(1, snapshot.SessionsClosed); + Assert.Equal(0, snapshot.Faults); + } + + /// + /// The sweep's counterpart of the kill race: a session that a concurrent close removed and + /// disposed between the snapshot and the close call is simply skipped — it is not a sweep + /// failure to log, and it is not counted as closed by the sweep. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CloseExpiredLeasesAsync_WhenCloseWonTheRace_DoesNotReportSweepFailure() + { + FakeWorkerClient workerClient = new(); + StaleRegistry registry = new(); + using GatewayMetrics metrics = new(); + SessionManager manager = CreateManager(new FakeSessionWorkerClientFactory(workerClient), registry, metrics); + GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + DateTimeOffset now = DateTimeOffset.UtcNow; + SetLeaseExpiry(session, now.AddSeconds(-1)); + + await manager.CloseSessionAsync(session.SessionId, CancellationToken.None); + registry.ServeStaleSnapshot(session); + + int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None); + + Assert.Equal(0, closedCount); + Assert.Equal(1, workerClient.ShutdownCount); + GatewayMetricsSnapshot snapshot = metrics.GetSnapshot(); + Assert.Equal(1, snapshot.SessionsClosed); + Assert.Equal(0, snapshot.OpenSessions); + Assert.Equal(0, snapshot.Faults); + } + /// /// Concurrent kills on the same session must not /// double-increment mxgateway.sessions.closed. The first kill wins, the second @@ -1086,6 +1151,40 @@ public sealed class SessionManagerTests Assert.Equal(1, registry.Count); } + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task OpenSessionAsync_WhenCallerCancelledDuringWorkerStartup_DiscardsWorkerClient() + { + FakeWorkerClient workerClient = new(); + GatedSessionWorkerClientFactory factory = new(workerClient); + SessionRegistry registry = new(); + using GatewayMetrics metrics = new(); + SessionManager manager = CreateManager(factory, registry, metrics); + using CancellationTokenSource cancellation = new(); + + Task openTask = manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, cancellation.Token); + GatewaySession starting = await factory.WaitForCreateStartAsync(); + + await cancellation.CancelAsync(); + factory.ReleaseCreate(); + SessionManagerException exception = await Assert.ThrowsAsync(async () => await openTask); + + Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode); + Assert.IsAssignableFrom(exception.InnerException); + Assert.Equal(1, workerClient.KillCount); + Assert.Equal(1, workerClient.DisposeCount); + Assert.NotEqual(SessionState.Ready, starting.State); + Assert.Equal(0, registry.Count); + Assert.Equal(0, metrics.GetSnapshot().OpenSessions); + Assert.Equal(0, metrics.GetSnapshot().SessionsOpened); + } + /// /// Same window as the kill case, driven by the host drain: a session still starting its /// worker when ShutdownAsync runs is closed by the drain, and the worker client that @@ -1296,6 +1395,66 @@ public sealed class SessionManagerTests subscriber.Dispose(); } + /// + /// The gateway's own alarm-monitor session only ever attaches an internal + /// (non-counted) subscriber. That must count as "attached" for this rule, or the gateway + /// would reap its own monitor every cycle once the knob is on. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CloseExpiredLeasesAsync_UnattachedGrace_DoesNotReapSessionWithInternalSubscriber() + { + FakeWorkerClient workerClient = new(); + FakeTimeProvider clock = new(DateTimeOffset.UtcNow); + SessionManager manager = CreateManager( + new FakeSessionWorkerClientFactory(workerClient), + options: CreateOptions(unattachedSessionGraceSeconds: 60), + timeProvider: clock); + GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + using IEventSubscriberLease internalLease = session.AttachInternalEventSubscriber(); + + clock.Advance(TimeSpan.FromSeconds(61)); + int closedCount = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None); + + Assert.Equal(0, closedCount); + Assert.Equal(SessionState.Ready, session.State); + Assert.Equal(0, workerClient.ShutdownCount); + } + + /// + /// The idle clock starts when the session becomes Ready, not when it was opened: a worker + /// that took longer than the window to start must not hand the client a session the very + /// next sweep reaps before it had a chance to subscribe. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CloseExpiredLeasesAsync_UnattachedGrace_IdleClockStartsAtReady() + { + FakeWorkerClient workerClient = new(); + GatedSessionWorkerClientFactory factory = new(workerClient); + FakeTimeProvider clock = new(DateTimeOffset.UtcNow); + SessionManager manager = CreateManager( + factory, + options: CreateOptions(unattachedSessionGraceSeconds: 60), + timeProvider: clock); + + Task openTask = manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None); + await factory.WaitForCreateStartAsync(); + clock.Advance(TimeSpan.FromSeconds(120)); + factory.ReleaseCreate(); + GatewaySession session = await openTask; + + int closedRightAfterReady = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None); + Assert.Equal(0, closedRightAfterReady); + Assert.Equal(SessionState.Ready, session.State); + + clock.Advance(TimeSpan.FromSeconds(60)); + int closedAfterIdle = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None); + + Assert.Equal(1, closedAfterIdle); + Assert.Equal(SessionState.Closed, session.State); + } + /// /// The knob is off by default: a never-subscribed idle session lives for its full lease. /// @@ -1878,12 +2037,81 @@ public sealed class SessionManagerTests GatewaySession session, CancellationToken cancellationToken) { + // Deliberately not bound to the caller's token: the tests that cancel the open while + // the worker is starting need the worker to still ARRIVE after the cancellation, which + // is the orphan-prone case. _entered.TrySetResult(session); - await _release.Task.WaitAsync(cancellationToken); + await _release.Task.WaitAsync(TimeSpan.FromSeconds(5)); return workerClient; } } + /// + /// Registry that can replay a race the real registry only produces by timing: a session a + /// concurrent close has already removed and disposed is still returned by one lookup (the + /// kill looked it up just before the removal) or still appears in one snapshot (the sweep + /// snapshotted just before the removal). + /// + private sealed class StaleRegistry : ISessionRegistry + { + private readonly SessionRegistry _inner = new(); + private GatewaySession? _staleLookup; + private GatewaySession? _staleSnapshot; + + /// Makes the next lookup of 's id return it although it was removed. + /// Removed session to serve once. + public void ServeStaleLookupOnce(GatewaySession session) + { + _staleLookup = session; + } + + /// Makes the next snapshot include although it was removed. + /// Removed session to include once. + public void ServeStaleSnapshot(GatewaySession session) + { + _staleSnapshot = session; + } + + /// + public int Count => _inner.Count; + + /// + public int ActiveCount => _inner.ActiveCount; + + /// + public bool TryAdd(GatewaySession session) => _inner.TryAdd(session); + + /// + public bool TryGet(string sessionId, [System.Diagnostics.CodeAnalysis.MaybeNullWhen(false)] out GatewaySession session) + { + if (_staleLookup is not null && _staleLookup.SessionId == sessionId) + { + session = _staleLookup; + _staleLookup = null; + return true; + } + + return _inner.TryGet(sessionId, out session); + } + + /// + public bool TryRemove(string sessionId, [System.Diagnostics.CodeAnalysis.MaybeNullWhen(false)] out GatewaySession session) + => _inner.TryRemove(sessionId, out session); + + /// + public IReadOnlyCollection Snapshot() + { + if (_staleSnapshot is null) + { + return _inner.Snapshot(); + } + + GatewaySession[] snapshot = [.. _inner.Snapshot(), _staleSnapshot]; + _staleSnapshot = null; + return snapshot; + } + } + /// /// Lifetime fake whose the test fires on demand; the /// other two tokens never fire.