fix(sessions): review follow-up — never orphan a late worker, exempt internal subscribers from unattached grace, tolerate lost close races

- Open path: the attach gate now waits on CancellationToken.None and the manager discards (kills + disposes) the late worker client on ANY exception out of the attach step, including the caller's own cancellation (checked first, so an abandoned open never yields a session). Previously an OperationCanceledException from the gate wait escaped before the discard and orphaned the x86 worker. The gate Release is guarded against ObjectDisposedException. - Unattached grace: AttachInternalEventSubscriber sets a separate latch so the gateway's own alarm-monitor session (internal subscriber only) is never reaped; the idle clock starts at Ready (MarkReady stamps _readyAtUtc) instead of at open, so a slow worker start does not eat the window. - KillWorkerAsync / CloseSessionCoreAsync: an ObjectDisposedException from the close gate on a session that is no longer registered means a concurrent close already removed, accounted and disposed it; both now return AlreadyClosed with no fault, no second gauge decrement and no sweep failure. The general kill catch only decrements the gauge when the session is still registered. The sweep no longer counts AlreadyClosed results among the sessions it closed.
2026-10-22 06:43:32 -04:00
parent 285ef64160
commit ec9f26400a
6 changed files with 346 additions and 33 deletions
@@ -63,11 +63,13 @@ public sealed class SessionOptions
    ///     and vanishes before it ever streams (for example OtOpcUa's <c>GalaxyDriver</c> failing
    ///     part-way through its start-up) otherwise pins a session slot and a live x86 worker for
    ///     the full <see cref="DefaultLeaseSeconds"/> (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 <see cref="DetachGraceSeconds"/> and the normal lease govern it). A value
-    ///     of <c>0</c> (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 <see cref="DetachGraceSeconds"/> and the normal lease govern
+    ///     it). A value of <c>0</c> (the default) disables the sweep. Must be greater than or
+    ///     equal to zero.
    /// </summary>
    /// <remarks>
    ///     Set it comfortably above the time a well-behaved client needs between
@@ -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 <see cref="OpenRecorded"/> is set right after it returns.
    /// </param>
-    /// <param name="cancellationToken">Token to cancel waiting for the gate.</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.
    /// </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.
+    /// </remarks>
    internal async ValueTask<bool> 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 —
@@ -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);
@@ -924,6 +924,71 @@ public sealed class SessionManagerTests
        Assert.True(metrics.GetSnapshot().Faults > 0);
    }

+    /// <summary>
+    ///     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 <c>AlreadyClosed</c> 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).
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
    /// <summary>
    /// Concurrent kills on the same session must not
    /// double-increment <c>mxgateway.sessions.closed</c>. The first kill wins, the second
@@ -1086,6 +1151,40 @@ public sealed class SessionManagerTests
        Assert.Equal(1, registry.Count);
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<GatewaySession> 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<SessionManagerException>(async () => await openTask);
+
+        Assert.Equal(SessionManagerErrorCode.OpenFailed, exception.ErrorCode);
+        Assert.IsAssignableFrom<OperationCanceledException>(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);
+    }
+
    /// <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
@@ -1296,6 +1395,66 @@ public sealed class SessionManagerTests
        subscriber.Dispose();
    }

+    /// <summary>
+    ///     The gateway's own alarm-monitor session only ever attaches an <em>internal</em>
+    ///     (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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<GatewaySession> 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);
+    }
+
    /// <summary>
    ///     The knob is off by default: a never-subscribed idle session lives for its full lease.
    /// </summary>
@@ -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;
        }
    }

+    /// <summary>
+    ///     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).
+    /// </summary>
+    private sealed class StaleRegistry : ISessionRegistry
+    {
+        private readonly SessionRegistry _inner = new();
+        private GatewaySession? _staleLookup;
+        private GatewaySession? _staleSnapshot;
+
+        /// <summary>Makes the next lookup of <paramref name="session"/>'s id return it although it was removed.</summary>
+        /// <param name="session">Removed session to serve once.</param>
+        public void ServeStaleLookupOnce(GatewaySession session)
+        {
+            _staleLookup = session;
+        }
+
+        /// <summary>Makes the next snapshot include <paramref name="session"/> although it was removed.</summary>
+        /// <param name="session">Removed session to include once.</param>
+        public void ServeStaleSnapshot(GatewaySession session)
+        {
+            _staleSnapshot = session;
+        }
+
+        /// <inheritdoc />
+        public int Count => _inner.Count;
+
+        /// <inheritdoc />
+        public int ActiveCount => _inner.ActiveCount;
+
+        /// <inheritdoc />
+        public bool TryAdd(GatewaySession session) => _inner.TryAdd(session);
+
+        /// <inheritdoc />
+        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);
+        }
+
+        /// <inheritdoc />
+        public bool TryRemove(string sessionId, [System.Diagnostics.CodeAnalysis.MaybeNullWhen(false)] out GatewaySession session)
+            => _inner.TryRemove(sessionId, out session);
+
+        /// <inheritdoc />
+        public IReadOnlyCollection<GatewaySession> Snapshot()
+        {
+            if (_staleSnapshot is null)
+            {
+                return _inner.Snapshot();
+            }
+
+            GatewaySession[] snapshot = [.. _inner.Snapshot(), _staleSnapshot];
+            _staleSnapshot = null;
+            return snapshot;
+        }
+    }
+
    /// <summary>
    ///     Lifetime fake whose <see cref="ApplicationStopping"/> the test fires on demand; the
    ///     other two tokens never fire.