mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs

using System.Runtime.CompilerServices;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Sessions;

/// <summary>
///     Concurrency and disposal regression tests for <see cref="GatewaySession"/>,
///     covering the split lock discipline between <c>_syncRoot</c> (state transitions)
///     and <c>_closeLock</c> (close serialization) and the gated <c>DisposeAsync</c>.
/// </summary>
public sealed class GatewaySessionTests
{
    /// <summary>
    ///     A <c>TransitionTo(Ready)</c> issued after
    ///     <see cref="GatewaySession.CloseAsync"/> has set <see cref="SessionState.Closing"/>
    ///     must not flip the session back to <see cref="SessionState.Ready"/>. The
    ///     blocking worker shutdown keeps <c>CloseAsync</c> parked between the
    ///     <c>Closing</c> write and the <c>Closed</c> write, which is precisely the
    ///     window the audit identified.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TransitionTo_AfterCloseStarted_DoesNotOverwriteClosing()
    {
        BlockingShutdownWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);

        Task<SessionCloseResult> closeTask = session.CloseAsync("test-close", CancellationToken.None);
        await workerClient.WaitForShutdownStartAsync();

        // Close has set _state = Closing under _syncRoot and is parked inside
        // worker.ShutdownAsync. A concurrent transition (e.g. a late
        // SessionWorkerClientFactory lifecycle callback) must not revive the session.
        Assert.Equal(SessionState.Closing, session.State);
        session.TransitionTo(SessionState.Ready);
        Assert.Equal(SessionState.Closing, session.State);

        workerClient.ReleaseShutdown();
        SessionCloseResult result = await closeTask;

        Assert.Equal(SessionState.Closed, result.FinalState);
        Assert.Equal(SessionState.Closed, session.State);

        await session.DisposeAsync();
    }

    /// <summary>
    ///     Once <see cref="GatewaySession.CloseAsync"/> finishes,
    ///     <see cref="GatewaySession.MarkFaulted"/> must not be able to move the
    ///     session out of <see cref="SessionState.Closed"/> either — the close path's
    ///     terminal write goes through the same <c>_syncRoot</c> the rest of the state
    ///     machine uses, so the existing "Closed is terminal" invariant holds.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task MarkFaulted_AfterCloseCompletes_DoesNotResurrectSession()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);

        await session.CloseAsync("test-close", CancellationToken.None);
        Assert.Equal(SessionState.Closed, session.State);

        session.MarkFaulted("late-fault");
        Assert.Equal(SessionState.Closed, session.State);

        await session.DisposeAsync();
    }

    /// <summary>
    ///     A <see cref="GatewaySession.MarkFaulted"/> issued
    ///     while <see cref="GatewaySession.CloseAsync"/> is parked between its
    ///     <c>Closing</c> and <c>Closed</c> writes must not break the close path's
    ///     terminal contract: the in-flight close runs to <c>Closed</c>, the fault
    ///     reason is preserved on <see cref="GatewaySession.FinalFault"/>, and the
    ///     session does not get stuck in <see cref="SessionState.Faulted"/>. The
    ///     state machine documents "Closing only allows a transition to Closed or
    ///     Faulted" — this test pins the resolved end state so a future tightening
    ///     of <c>MarkFaulted</c> cannot silently regress it.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task MarkFaulted_DuringInFlightClose_PreservesFaultButYieldsToClose()
    {
        BlockingShutdownWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);

        Task<SessionCloseResult> closeTask = session.CloseAsync("test-close", CancellationToken.None);
        await workerClient.WaitForShutdownStartAsync();

        // Close has set _state = Closing under _syncRoot and is parked inside
        // worker.ShutdownAsync. Fault the session from another thread while parked.
        Assert.Equal(SessionState.Closing, session.State);
        session.MarkFaulted("concurrent-fault");

        workerClient.ReleaseShutdown();
        SessionCloseResult result = await closeTask;

        // Close still wins — Closed is terminal — but the fault reason is preserved
        // so observers see the original cause once the session settles.
        Assert.Equal(SessionState.Closed, result.FinalState);
        Assert.Equal(SessionState.Closed, session.State);
        Assert.Equal("concurrent-fault", session.FinalFault);

        await session.DisposeAsync();
    }

    /// <summary>
    ///     <see cref="GatewaySession.DisposeAsync"/> must wait
    ///     for an in-flight <see cref="GatewaySession.CloseAsync"/> before disposing
    ///     its semaphore. Without the fix, the close's <c>_closeLock.Release()</c>
    ///     would race the dispose and raise <see cref="ObjectDisposedException"/>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DisposeAsync_WhileCloseInFlight_WaitsForCloseAndDoesNotThrow()
    {
        BlockingShutdownWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);

        Task<SessionCloseResult> closeTask = session.CloseAsync("test-close", CancellationToken.None);
        await workerClient.WaitForShutdownStartAsync();

        // Start disposing while close is still parked inside worker.ShutdownAsync.
        ValueTask disposeTask = session.DisposeAsync();

        // Now release the worker shutdown so close can complete.
        workerClient.ReleaseShutdown();

        // Both must complete cleanly — the close's Release() must run before the
        // dispose actually tears the semaphore down.
        SessionCloseResult result = await closeTask;
        await disposeTask;

        Assert.Equal(SessionState.Closed, result.FinalState);
        Assert.Equal(1, workerClient.ShutdownCount);
        // Worker dispose ran exactly once even with the close/dispose interleave.
        Assert.Equal(1, workerClient.DisposeCount);
    }

    /// <summary>
    ///     Double-dispose is tolerated: the second call must swallow
    ///     <see cref="ObjectDisposedException"/> from the already-disposed semaphore
    ///     rather than propagating it.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DisposeAsync_CalledTwice_DoesNotThrow()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySession(workerClient);
        await session.CloseAsync("test-close", CancellationToken.None);

        await session.DisposeAsync();
        // No second exception — the dispose's defensive ObjectDisposedException catch
        // covers the doubled call path that SessionManager.ShutdownAsync could trigger
        // if it re-removed a session.
        await session.DisposeAsync();
    }

    /// <summary>
    ///     Concurrent <c>Dispose()</c> calls on the same
    ///     <see cref="IEventSubscriberLease"/> — as can happen when a gRPC stream
    ///     completion and a client cancellation both fire at the same time — must
    ///     decrement <c>_activeEventSubscriberCount</c> exactly once, never to −1.
    ///     A negative count permanently blocks future subscribers because
    ///     <c>AttachEventSubscriber</c> gates on <c>_activeEventSubscriberCount >= effectiveCap</c>.
    ///     After both racing disposes finish,
    ///     the count must be exactly 0 and a subsequent single-subscriber attach must
    ///     succeed.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task EventSubscriberLease_ConcurrentDispose_DecrementsCountExactlyOnce()
    {
        const int Concurrency = 16;
        const int Iterations = 200;
        TimeSpan testTimeout = TimeSpan.FromSeconds(10);

        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(workerClient);

        for (int i = 0; i < Iterations; i++)
        {
            // Attach one subscriber; this increments _activeEventSubscriberCount to 1.
            IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);

            // Race Concurrency threads all calling Dispose() on the same lease.
            // Only one must actually run DetachEventSubscriber.
            using SemaphoreSlim gate = new(0);
            Task[] tasks = new Task[Concurrency];
            for (int t = 0; t < Concurrency; t++)
            {
                tasks[t] = Task.Run(async () =>
                {
                    // All threads wait at the gate so they start as simultaneously
                    // as the scheduler allows, maximising the race window.
                    await gate.WaitAsync(testTimeout);
                    lease.Dispose();
                });
            }

            gate.Release(Concurrency);
            await Task.WhenAll(tasks).WaitAsync(testTimeout);

            // Count must be exactly 0 — not negative — after all disposes.
            Assert.Equal(0, session.ActiveEventSubscriberCount);

            // Observable contract: a fresh single subscriber must now be attachable
            // (i.e., the guard _activeEventSubscriberCount > 0 is false).
            IEventSubscriberLease next = session.AttachEventSubscriber(maxSubscribers: 1);
            next.Dispose();
            Assert.Equal(0, session.ActiveEventSubscriberCount);
        }

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     Single-subscriber mode rejects a SECOND concurrent external
    ///     attach with <see cref="SessionManagerErrorCode.EventSubscriberAlreadyActive"/> —
    ///     the legacy guard is preserved unchanged when multi-subscriber is disabled.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachEventSubscriber_SingleMode_SecondAttachThrowsAlreadyActive()
    {
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(workerClient);

        using IEventSubscriberLease first = session.AttachEventSubscriber(maxSubscribers: 8);

        SessionManagerException exception = Assert.Throws<SessionManagerException>(
            () => session.AttachEventSubscriber(maxSubscribers: 8));
        Assert.Equal(SessionManagerErrorCode.EventSubscriberAlreadyActive, exception.ErrorCode);
        Assert.Equal(1, session.ActiveEventSubscriberCount);

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     Multi-subscriber mode allows exactly <c>cap</c> concurrent external
    ///     subscribers; the (cap+1)-th attach throws
    ///     <see cref="SessionManagerErrorCode.EventSubscriberLimitReached"/>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachEventSubscriber_MultiMode_AttachesUpToCapThenThrowsLimitReached()
    {
        const int Cap = 4;
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(
            workerClient,
            allowMultipleEventSubscribers: true);

        List<IEventSubscriberLease> leases = [];
        for (int i = 0; i < Cap; i++)
        {
            leases.Add(session.AttachEventSubscriber(maxSubscribers: Cap));
        }

        Assert.Equal(Cap, session.ActiveEventSubscriberCount);

        SessionManagerException exception = Assert.Throws<SessionManagerException>(
            () => session.AttachEventSubscriber(maxSubscribers: Cap));
        Assert.Equal(SessionManagerErrorCode.EventSubscriberLimitReached, exception.ErrorCode);
        Assert.Equal(Cap, session.ActiveEventSubscriberCount);

        foreach (IEventSubscriberLease lease in leases)
        {
            lease.Dispose();
        }

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     The gateway-owned INTERNAL dashboard subscriber must NOT consume cap
    ///     budget: with the dashboard mirror running, the full cap of external subscribers is
    ///     still attachable.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachEventSubscriber_MultiMode_DashboardMirrorDoesNotConsumeCap()
    {
        const int Cap = 3;
        FakeWorkerClient workerClient = new();
        RecordingDashboardEventBroadcaster broadcaster = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(
            workerClient,
            allowMultipleEventSubscribers: true,
            dashboardBroadcaster: broadcaster);

        // The internal dashboard mirror registered on MarkReady is NOT counted as an external
        // subscriber, so the external active count starts at zero.
        Assert.Equal(0, session.ActiveEventSubscriberCount);

        List<IEventSubscriberLease> leases = [];
        for (int i = 0; i < Cap; i++)
        {
            leases.Add(session.AttachEventSubscriber(maxSubscribers: Cap));
        }

        Assert.Equal(Cap, session.ActiveEventSubscriberCount);

        // The (cap+1)-th still fails: the dashboard mirror did not eat a slot.
        SessionManagerException exception = Assert.Throws<SessionManagerException>(
            () => session.AttachEventSubscriber(maxSubscribers: Cap));
        Assert.Equal(SessionManagerErrorCode.EventSubscriberLimitReached, exception.ErrorCode);

        foreach (IEventSubscriberLease lease in leases)
        {
            lease.Dispose();
        }

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     Many concurrent attaches in multi-subscriber mode must never
    ///     exceed the cap: the count-check-and-increment is atomic under <c>_syncRoot</c>, so
    ///     exactly <c>cap</c> attaches succeed and the rest throw
    ///     <see cref="SessionManagerErrorCode.EventSubscriberLimitReached"/>. The observed
    ///     count never goes above the cap.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachEventSubscriber_MultiMode_ConcurrentAttaches_NeverExceedCap()
    {
        const int Cap = 5;
        const int Attempts = 32;
        TimeSpan testTimeout = TimeSpan.FromSeconds(10);

        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(
            workerClient,
            allowMultipleEventSubscribers: true);

        using SemaphoreSlim gate = new(0);
        int successCount = 0;
        int limitReachedCount = 0;
        int maxObservedCount = 0;
        IEventSubscriberLease?[] leases = new IEventSubscriberLease?[Attempts];
        Task[] tasks = new Task[Attempts];
        for (int t = 0; t < Attempts; t++)
        {
            int index = t;
            tasks[index] = Task.Run(async () =>
            {
                await gate.WaitAsync(testTimeout);
                try
                {
                    IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: Cap);
                    leases[index] = lease;
                    Interlocked.Increment(ref successCount);
                }
                catch (SessionManagerException exception)
                    when (exception.ErrorCode == SessionManagerErrorCode.EventSubscriberLimitReached)
                {
                    Interlocked.Increment(ref limitReachedCount);
                }

                int observed = session.ActiveEventSubscriberCount;
                int previousMax;
                do
                {
                    previousMax = Volatile.Read(ref maxObservedCount);
                    if (observed <= previousMax)
                    {
                        break;
                    }
                }
                while (Interlocked.CompareExchange(ref maxObservedCount, observed, previousMax) != previousMax);
            });
        }

        gate.Release(Attempts);
        await Task.WhenAll(tasks).WaitAsync(testTimeout);

        Assert.Equal(Cap, successCount);
        Assert.Equal(Attempts - Cap, limitReachedCount);
        Assert.Equal(Cap, session.ActiveEventSubscriberCount);
        Assert.True(maxObservedCount <= Cap, $"Observed count {maxObservedCount} exceeded cap {Cap}.");

        foreach (IEventSubscriberLease? lease in leases)
        {
            lease?.Dispose();
        }

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     Disposing a subscriber frees a cap slot so a fresh attach succeeds, and a
    ///     double-dispose does not double-free the slot (count integrity preserved).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachEventSubscriber_MultiMode_DisposeFreesSlotAndDoubleDisposeIsIdempotent()
    {
        const int Cap = 2;
        FakeWorkerClient workerClient = new();
        GatewaySession session = CreateReadySessionWithEventStreaming(
            workerClient,
            allowMultipleEventSubscribers: true);

        IEventSubscriberLease a = session.AttachEventSubscriber(maxSubscribers: Cap);
        IEventSubscriberLease b = session.AttachEventSubscriber(maxSubscribers: Cap);
        Assert.Equal(Cap, session.ActiveEventSubscriberCount);

        // At cap: next attach is rejected.
        Assert.Throws<SessionManagerException>(
            () => session.AttachEventSubscriber(maxSubscribers: Cap));

        // Dispose one — and dispose it twice. The second dispose must not double-free.
        a.Dispose();
        a.Dispose();
        Assert.Equal(1, session.ActiveEventSubscriberCount);

        // Exactly one slot is free, so exactly one fresh attach succeeds.
        using IEventSubscriberLease c = session.AttachEventSubscriber(maxSubscribers: Cap);
        Assert.Equal(Cap, session.ActiveEventSubscriberCount);
        Assert.Throws<SessionManagerException>(
            () => session.AttachEventSubscriber(maxSubscribers: Cap));

        b.Dispose();

        await session.CloseAsync("test-done", CancellationToken.None);
        await session.DisposeAsync();
    }

    /// <summary>
    ///     With a positive detach-grace, dropping the last external subscriber must
    ///     RETAIN the session (it stays <see cref="SessionState.Ready"/>, not Closed/Faulted)
    ///     and record a detached timestamp so the lease monitor can age it out later.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_LastSubscriberDrops_RetainsSessionAndRecordsDetachedTimestamp()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30));

        IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
        Assert.Null(session.DetachedAtUtc);

        lease.Dispose();

        // Retained, not torn down.
        Assert.Equal(SessionState.Ready, session.State);
        Assert.Equal(0, session.ActiveEventSubscriberCount);
        Assert.Equal(clock.GetUtcNow(), session.DetachedAtUtc);
        Assert.False(session.IsDetachGraceExpired(clock.GetUtcNow()));
    }

    /// <summary>
    ///     Advancing the clock past the detach-grace window makes the retained,
    ///     detached session eligible for close (<see cref="GatewaySession.IsDetachGraceExpired"/>).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_ClockAdvancesPastWindow_SessionBecomesEligibleForClose()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30));

        IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
        lease.Dispose();

        // Just before the window elapses: not yet eligible.
        clock.Advance(TimeSpan.FromSeconds(29));
        Assert.False(session.IsDetachGraceExpired(clock.GetUtcNow()));

        // At/after the window: eligible for close.
        clock.Advance(TimeSpan.FromSeconds(1));
        Assert.True(session.IsDetachGraceExpired(clock.GetUtcNow()));
    }

    /// <summary>
    ///     Re-attaching a subscriber before the window elapses cancels the grace:
    ///     the detached timestamp clears and a subsequent clock advance does NOT make the
    ///     session eligible for close.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_ReattachBeforeExpiry_CancelsGrace()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30));

        IEventSubscriberLease first = session.AttachEventSubscriber(maxSubscribers: 1);
        first.Dispose();
        Assert.NotNull(session.DetachedAtUtc);

        clock.Advance(TimeSpan.FromSeconds(10));
        IEventSubscriberLease second = session.AttachEventSubscriber(maxSubscribers: 1);

        // Re-attach cancelled the grace window.
        Assert.Null(session.DetachedAtUtc);

        // Advancing well past what would have been the window does not make it eligible while
        // a subscriber is attached.
        clock.Advance(TimeSpan.FromMinutes(5));
        Assert.False(session.IsDetachGraceExpired(clock.GetUtcNow()));

        second.Dispose();
    }

    /// <summary>
    ///     With detach-grace disabled (<c>0</c>), dropping the last subscriber must
    ///     match today's behavior: the session stays <see cref="SessionState.Ready"/> with no
    ///     detached timestamp and is never eligible for a detach-grace close — it lingers only
    ///     until its normal lease expires.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_Disabled_MatchesTodaysBehavior()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.Zero);

        IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
        lease.Dispose();

        Assert.Equal(SessionState.Ready, session.State);
        Assert.Null(session.DetachedAtUtc);

        clock.Advance(TimeSpan.FromHours(1));
        Assert.False(session.IsDetachGraceExpired(clock.GetUtcNow()));
    }

    /// <summary>
    ///     A FAILED first attach (the distributor never registered a
    ///     subscriber) must NOT enter the detach-grace window: the catch path's
    ///     <c>DetachEventSubscriber</c> rolls the reserved slot back to 0 but must not stamp
    ///     <c>DetachedAtUtc</c>, because the "last subscriber dropped" semantics only apply once
    ///     a subscriber was successfully registered. A freshly-Ready session whose first attach
    ///     failed must therefore stay out of grace and never become sweep-eligible on that basis.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_FailedFirstAttach_DoesNotEnterGrace()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();

        // QueueCapacity = 0 makes the distributor constructor throw ArgumentOutOfRangeException
        // inside StartDistributorAndRegister, so the very first AttachEventSubscriber fails after
        // it reserved a slot — exercising the catch → DetachEventSubscriber rollback path.
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30),
            queueCapacity: 0);

        Assert.ThrowsAny<ArgumentException>(
            () => session.AttachEventSubscriber(maxSubscribers: 1));

        // The reserved slot was rolled back, but no successful subscriber ever existed, so the
        // session must NOT have entered detach-grace.
        Assert.Equal(SessionState.Ready, session.State);
        Assert.Equal(0, session.ActiveEventSubscriberCount);
        Assert.Null(session.DetachedAtUtc);

        // And it must never become detach-grace-eligible no matter how far the clock advances.
        clock.Advance(TimeSpan.FromHours(1));
        Assert.False(session.IsDetachGraceExpired(clock.GetUtcNow()));
    }

    /// <summary>
    ///     The gateway-owned internal dashboard subscriber must NOT keep a session out
    ///     of detach-grace: with only the dashboard mirror attached (and no external gRPC
    ///     subscriber), dropping the last external subscriber still enters grace and the
    ///     window still expires.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DetachGrace_DashboardMirrorAlone_DoesNotPreventGraceEntry()
    {
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        FakeWorkerClient workerClient = new();
        RecordingDashboardEventBroadcaster broadcaster = new();
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30),
            dashboardBroadcaster: broadcaster);

        // The dashboard mirror is the only subscriber (registered internally at MarkReady).
        // It is not counted as an external subscriber.
        Assert.Equal(0, session.ActiveEventSubscriberCount);

        IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
        lease.Dispose();

        // Entered grace despite the dashboard mirror still being attached.
        Assert.NotNull(session.DetachedAtUtc);

        clock.Advance(TimeSpan.FromSeconds(30));
        Assert.True(session.IsDetachGraceExpired(clock.GetUtcNow()));
    }

    /// <summary>
    ///     Validates the TOCTOU fix: TryBeginCloseIfExpired atomically re-checks that no
    ///     subscriber has reattached before flipping to Closing. When the grace window has elapsed but
    ///     a subscriber is attached by the time TryBeginCloseIfExpired runs, it returns false and the
    ///     session remains Ready.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task TryBeginCloseIfExpired_ReattachedSubscriberWinsRace_DeclinesClose()
    {
        FakeWorkerClient workerClient = new();
        FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
        await using GatewaySession session = CreateReadySessionWithDetachGrace(
            workerClient,
            clock,
            detachGrace: TimeSpan.FromSeconds(30));

        // Attach then drop to enter detach-grace, then advance past the window.
        IDisposable firstSubscriber = session.AttachEventSubscriber(maxSubscribers: 1);
        firstSubscriber.Dispose();
        clock.Advance(TimeSpan.FromSeconds(31));
        DateTimeOffset expiredNow = clock.GetUtcNow();
        Assert.True(session.IsDetachGraceExpired(expiredNow)); // sanity: would be closed by sweep

        // Simulate a client reconnecting before the sweeper calls TryBeginCloseIfExpired.
        // The reattach clears _detachedAtUtc and increments _activeEventSubscriberCount so
        // neither expiry condition holds any longer.
        using IDisposable reconnected = session.AttachEventSubscriber(maxSubscribers: 1);
        Assert.Null(session.DetachedAtUtc);

        // TryBeginCloseIfExpired must see the reattach and decline — the session stays Ready.
        bool began = session.TryBeginCloseIfExpired(expiredNow, out bool alreadyClosing);

        Assert.False(began);
        Assert.False(alreadyClosing);
        Assert.Equal(SessionState.Ready, session.State);
    }

    private static GatewaySession CreateReadySessionWithDetachGrace(
        IWorkerClient workerClient,
        TimeProvider timeProvider,
        TimeSpan detachGrace,
        IDashboardEventBroadcaster? dashboardBroadcaster = null,
        int queueCapacity = 8)
    {
        GatewaySession session = new(
            sessionId: "session-test-detach-grace",
            backendName: "mxaccess",
            pipeName: "mxaccess-gateway-1-session-test-detach-grace",
            nonce: "nonce",
            clientIdentity: "client-1",
            ownerKeyId: null,
            clientSessionName: "test-session",
            clientCorrelationId: "client-correlation-1",
            commandTimeout: TimeSpan.FromSeconds(5),
            startupTimeout: TimeSpan.FromSeconds(5),
            shutdownTimeout: TimeSpan.FromSeconds(5),
            leaseDuration: TimeSpan.FromMinutes(30),
            openedAt: timeProvider.GetUtcNow(),
            eventStreaming: new SessionEventStreaming(
                new MxAccessGrpcMapper(),
                new EventOptions { QueueCapacity = queueCapacity },
                NullLogger<SessionEventDistributor>.Instance,
                timeProvider,
                new GatewayMetrics(),
                dashboardBroadcaster),
            detachGrace: detachGrace);
        session.AttachWorkerClient(workerClient);
        session.MarkReady();
        return session;
    }

    private static GatewaySession CreateReadySession(IWorkerClient workerClient)
    {
        GatewaySession session = new(
            sessionId: "session-test",
            backendName: "mxaccess",
            pipeName: "mxaccess-gateway-1-session-test",
            nonce: "nonce",
            clientIdentity: "client-1",
            ownerKeyId: null,
            clientSessionName: "test-session",
            clientCorrelationId: "client-correlation-1",
            commandTimeout: TimeSpan.FromSeconds(5),
            startupTimeout: TimeSpan.FromSeconds(5),
            shutdownTimeout: TimeSpan.FromSeconds(5),
            leaseDuration: TimeSpan.FromMinutes(30),
            openedAt: DateTimeOffset.UtcNow);
        session.AttachWorkerClient(workerClient);
        session.MarkReady();
        return session;
    }

    private static GatewaySession CreateReadySessionWithEventStreaming(
        IWorkerClient workerClient,
        bool allowMultipleEventSubscribers = false,
        IDashboardEventBroadcaster? dashboardBroadcaster = null)
    {
        GatewaySession session = new(
            sessionId: "session-test-concurrent",
            backendName: "mxaccess",
            pipeName: "mxaccess-gateway-1-session-test-concurrent",
            nonce: "nonce",
            clientIdentity: "client-1",
            ownerKeyId: null,
            clientSessionName: "test-session",
            clientCorrelationId: "client-correlation-1",
            commandTimeout: TimeSpan.FromSeconds(5),
            startupTimeout: TimeSpan.FromSeconds(5),
            shutdownTimeout: TimeSpan.FromSeconds(5),
            leaseDuration: TimeSpan.FromMinutes(30),
            openedAt: DateTimeOffset.UtcNow,
            eventStreaming: new SessionEventStreaming(
                new MxAccessGrpcMapper(),
                new EventOptions { QueueCapacity = 8 },
                NullLogger<SessionEventDistributor>.Instance,
                TimeProvider.System,
                new GatewayMetrics(),
                dashboardBroadcaster,
                allowMultipleEventSubscribers));
        session.AttachWorkerClient(workerClient);
        session.MarkReady();
        return session;
    }

    /// <summary>
    ///     Minimal worker client that parks <see cref="ShutdownAsync"/> until the test
    ///     explicitly releases it. Used to keep <see cref="GatewaySession.CloseAsync"/>
    ///     stuck between its <c>Closing</c> and <c>Closed</c> writes so the test can
    ///     observe and act on the intermediate state.
    /// </summary>
    private sealed class BlockingShutdownWorkerClient : IWorkerClient
    {
        private readonly TaskCompletionSource _shutdownStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly TaskCompletionSource _shutdownReleased = new(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <inheritdoc />
        public string SessionId { get; } = "session-test";

        /// <inheritdoc />
        public int? ProcessId { get; } = 1234;

        /// <inheritdoc />
        public WorkerClientState State { get; private set; } = WorkerClientState.Ready;

        /// <inheritdoc />
        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;

        /// <summary>Gets the count of shutdown invocations.</summary>
        public int ShutdownCount { get; private set; }

        /// <summary>Gets the count of dispose invocations.</summary>
        public int DisposeCount { get; private set; }

        /// <summary>Waits for shutdown to start.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public Task WaitForShutdownStartAsync()
        {
            return _shutdownStarted.Task.WaitAsync(TimeSpan.FromSeconds(5));
        }

        /// <summary>Releases the shutdown block.</summary>
        public void ReleaseShutdown()
        {
            _shutdownReleased.TrySetResult();
        }

        /// <inheritdoc />
        public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            WorkerCommand command,
            TimeSpan timeout,
            CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());

        /// <inheritdoc />
        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await Task.CompletedTask.ConfigureAwait(false);
            yield break;
        }

        /// <inheritdoc />
        public async Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken)
        {
            ShutdownCount++;
            _shutdownStarted.TrySetResult();
            await _shutdownReleased.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
            State = WorkerClientState.Closed;
        }

        /// <inheritdoc />
        public void Kill(string reason)
        {
            State = WorkerClientState.Faulted;
        }

        /// <summary>Releases the fake worker resources and records the invocation.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public ValueTask DisposeAsync()
        {
            DisposeCount++;
            return ValueTask.CompletedTask;
        }
    }

    private sealed class FakeWorkerClient : IWorkerClient
    {
        /// <inheritdoc />
        public string SessionId { get; } = "session-test";

        /// <inheritdoc />
        public int? ProcessId { get; } = 1234;

        /// <inheritdoc />
        public WorkerClientState State { get; } = WorkerClientState.Ready;

        /// <inheritdoc />
        public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;

        /// <summary>Gets the count of dispose invocations.</summary>
        public int DisposeCount { get; private set; }

        /// <inheritdoc />
        public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            WorkerCommand command,
            TimeSpan timeout,
            CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());

        /// <inheritdoc />
        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await Task.CompletedTask.ConfigureAwait(false);
            yield break;
        }

        /// <inheritdoc />
        public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;

        /// <inheritdoc />
        public void Kill(string reason)
        {
        }

        /// <summary>Releases the fake worker resources and records the invocation.</summary>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public ValueTask DisposeAsync()
        {
            DisposeCount++;
            return ValueTask.CompletedTask;
        }
    }
}