fix(events): graceful unregister no longer masquerades as overflow under FailFast

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Time.Testing;
@@ -1114,6 +1115,96 @@ public sealed class SessionEventDistributorTests
         Assert.Equal(1, distributor.SubscriberCount);
     }
 
+    /// <summary>
+    ///     Regression: a subscriber that unregisters (lease disposed) after the pump captured the
+    ///     fan-out array is still written to, and <c>TryWrite</c> on its now-completed channel
+    ///     returns false — the same signal a full channel gives. Treating that as backpressure
+    ///     emitted a bogus <c>EventQueueOverflow</c> metric and, under the default
+    ///     single-subscriber FailFast policy, faulted the whole session: a stream ending normally
+    ///     during traffic could kill the session. The overflow path must claim the removal first
+    ///     and bail out when the subscriber is already gone.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task GracefulUnregisterDuringFanOut_DoesNotReportOverflow_OrFaultTheSession()
+    {
+        Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
+
+        // Records every overflow-handler invocation. isInternal distinguishes the deliberate
+        // overflow (the internal subscriber below) from the graceful unregister under test.
+        ConcurrentQueue<(bool IsOnlySubscriber, bool IsInternal)> invocations = new();
+        IEventSubscriberLease? gracefulLease = null;
+        int disposedGracefulLease = 0;
+
+        await using SessionEventDistributor distributor = new(
+            "session-graceful-unregister",
+            ct => source.Reader.ReadAllAsync(ct),
+            subscriberQueueCapacity: 1,
+            replayBufferCapacity: 0,
+            replayRetentionSeconds: 0,
+            NullLogger<SessionEventDistributor>.Instance,
+            TimeProvider.System,
+            (isOnlySubscriber, isInternal) =>
+            {
+                invocations.Enqueue((isOnlySubscriber, isInternal));
+
+                // The seam that makes the race deterministic: this handler runs ON the pump
+                // thread, part-way through fanning one event to the array it already captured.
+                // Disposing the graceful lease here unregisters and completes that subscriber
+                // in exactly the window the fix targets — after the capture, before the pump
+                // reaches its TryWrite. Only on the first invocation, so a genuine repeat
+                // overflow cannot re-trigger it.
+                if (Interlocked.Exchange(ref disposedGracefulLease, 1) == 0)
+                {
+                    gracefulLease!.Dispose();
+                }
+            },
+            singleSubscriberMode: true);
+
+        await distributor.StartAsync(CancellationToken.None);
+
+        // Registered FIRST so it precedes the graceful subscriber in the captured fan-out array,
+        // putting the graceful subscriber's TryWrite after this one's overflow handler. Internal
+        // so its own (expected) overflow reports isOnlySubscriber == false and can never fault
+        // the session by itself. Never read from, so its capacity-1 channel fills immediately.
+        using IEventSubscriberLease overflowing = distributor.Register(isInternal: true);
+
+        // External subscriber that will unregister gracefully mid-fan-out. Under the old
+        // behavior its completed-channel TryWrite reported isOnlySubscriber == true, which is
+        // precisely the legacy FailFast "fault the session" signal.
+        gracefulLease = distributor.Register();
+
+        // Event 1 fills the internal subscriber's channel and is drained from the graceful one,
+        // so on event 2 the internal subscriber overflows while the graceful one has room —
+        // whichever order the array happens to hold, only the internal subscriber overflows.
+        source.Writer.TryWrite(Event(1));
+        MxEvent first = await ReadOneAsync(gracefulLease.Reader);
+        Assert.Equal(1ul, first.WorkerSequence);
+
+        // Event 2: the internal subscriber overflows, the handler disposes the graceful lease,
+        // and the pump then writes event 2 to that already-completed channel.
+        source.Writer.TryWrite(Event(2));
+
+        // The graceful subscriber's channel must complete cleanly — no EventQueueOverflow fault.
+        await AssertCompletedAsync(gracefulLease.Reader);
+
+        // The pump survives and keeps serving a freshly-attached subscriber.
+        using IEventSubscriberLease later = distributor.Register();
+        source.Writer.TryWrite(Event(3));
+        Assert.Equal(3ul, (await ReadOneAsync(later.Reader)).WorkerSequence);
+
+        // Guards against a vacuous pass: the deliberate internal overflow must actually have
+        // fired, since that handler call is the seam that disposes the lease mid-fan-out.
+        Assert.NotEmpty(invocations);
+        Assert.Equal(1, Volatile.Read(ref disposedGracefulLease));
+
+        // The deliberate internal overflow is expected; the graceful unregister must NOT have
+        // produced an overflow report of its own. An isOnlySubscriber == true invocation is the
+        // exact signal that would have faulted the session.
+        Assert.All(invocations, invocation => Assert.True(invocation.IsInternal));
+        Assert.DoesNotContain(invocations, invocation => invocation.IsOnlySubscriber);
+    }
+
     private static SessionEventDistributor CreateDistributor(ChannelReader<MxEvent> source)
         => CreateDistributor(source, replayBufferCapacity: 1024, replayRetentionSeconds: 300);