diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs index 9ea3ec4..ba97a03 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs @@ -683,14 +683,35 @@ public sealed class SessionEventDistributor : IAsyncDisposable } // Applies the per-subscriber backpressure policy when a subscriber's bounded channel is - // full. Runs on the pump thread. The offending subscriber is ALWAYS disconnected with an - // overflow fault and unregistered, so it can never wedge the pump again; the overflow - // handler decides the observable side effects (overflow metric, and — for legacy + // full — or, indistinguishably from the pump's side, already completed. A subscriber that + // really overflowed is ALWAYS disconnected with an overflow fault and unregistered, so it + // can never wedge the pump again; one that merely unregistered itself is dropped silently + // (see the discriminator below). Runs on the pump thread. The overflow handler decides the + // observable side effects (overflow metric, and — for legacy // single-subscriber FailFast — faulting the owning session). Multi-subscriber FailFast // intentionally degrades to a plain disconnect (see SubscriberOverflowHandler docs): one // slow consumer must not fault a session shared by other healthy subscribers. private void OnSubscriberOverflow(Subscriber subscriber, ulong workerSequence) { + // Claim the disconnect FIRST, because a false TryWrite is ambiguous. It means either + // "channel full" (a genuine overflow) or "channel already completed" — which happens + // when the subscriber unregistered after the pump captured the fan-out array and is + // therefore a GRACEFUL close, not backpressure. RemoveSubscriber separates the two: + // every path that completes a channel during fan-out (lease disposal via Unregister, + // and this method) removes the subscriber from the set BEFORE completing it, so a + // completed channel implies the subscriber is already gone and RemoveSubscriber + // returns false. (CompleteAllSubscribers completes without removing, but only after + // the pump has left its loop, so it cannot be observed here.) + // + // Bailing out on false is what keeps a normal stream ending mid-traffic from emitting + // a bogus EventQueueOverflow metric and — under the default single-subscriber FailFast + // policy — faulting the whole session. Winning the removal also guarantees the side + // effects below run exactly once per subscriber. + if (!RemoveSubscriber(subscriber)) + { + return; + } + // Decide whether FailFast may fault the whole session for this overflow. This is the // "isOnlySubscriber" signal the legacy single-subscriber FailFast path keys on. bool isOnlySubscriber = !subscriber.IsInternal && _singleSubscriberMode; @@ -717,20 +738,15 @@ public sealed class SessionEventDistributor : IAsyncDisposable subscriber.Id); } - // Disconnect ONLY this subscriber: complete its channel with the overflow fault and - // remove it from the fan-out set. Its gRPC reader's MoveNextAsync then throws the - // SessionManagerException, which EventStreamService surfaces to the client exactly as - // the pre-epic per-RPC overflow did. The pump and every other subscriber are untouched. - // - // The removal takes _lifecycleLock because it must republish the fan-out snapshot; the - // pump holds no other lock here (fan-out runs outside _replayLock), so this cannot - // invert the _replayLock-then-_lifecycleLock order RegisterWithReplay uses. - if (RemoveSubscriber(subscriber)) - { - subscriber.Channel.Writer.TryComplete(new SessionManagerException( - SessionManagerErrorCode.EventQueueOverflow, - $"Session {_sessionId} event stream queue overflowed.")); - } + // Disconnect ONLY this subscriber: it is already out of the fan-out set (removed above), + // so complete its channel with the overflow fault. Its gRPC reader's MoveNextAsync then + // throws the SessionManagerException, which EventStreamService surfaces to the client + // exactly as the pre-epic per-RPC overflow did. The pump and every other subscriber are + // untouched. This runs even when the handler above threw — the subscriber must never be + // left attached with an un-completed channel. + subscriber.Channel.Writer.TryComplete(new SessionManagerException( + SessionManagerErrorCode.EventQueueOverflow, + $"Session {_sessionId} event stream queue overflowed.")); } private void CompleteAllSubscribers(Exception? error) @@ -765,6 +781,11 @@ public sealed class SessionEventDistributor : IAsyncDisposable // Returns true only for the caller that actually removed it, so the channel is completed // exactly once however many disposal/overflow paths race. Completing the channel is left to // that caller and happens OUTSIDE the lock: this lock guards set membership only. + // + // Remove-then-complete (never the reverse) is load-bearing, not incidental: it is what lets + // OnSubscriberOverflow read a false return as "this subscriber unregistered gracefully" + // rather than "this subscriber overflowed". Completing before removing would resurrect the + // spurious-session-fault bug. private bool RemoveSubscriber(Subscriber subscriber) { lock (_lifecycleLock) diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs index 75a74ad..1338a71 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs +++ b/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); } + /// + /// Regression: a subscriber that unregisters (lease disposed) after the pump captured the + /// fan-out array is still written to, and TryWrite on its now-completed channel + /// returns false — the same signal a full channel gives. Treating that as backpressure + /// emitted a bogus EventQueueOverflow 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. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task GracefulUnregisterDuringFanOut_DoesNotReportOverflow_OrFaultTheSession() + { + Channel source = Channel.CreateUnbounded(); + + // 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.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 source) => CreateDistributor(source, replayBufferCapacity: 1024, replayRetentionSeconds: 300);