diff --git a/docs/Sessions.md b/docs/Sessions.md index 77f8621..ce2d997 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -213,6 +213,8 @@ The repair transitions the monitor's reconcile broadcasts on the alarm feed (Rai `AttachInternalEventSubscriber` enforces the same readiness gate as `AttachEventSubscriber` — a session (or worker) that is not `Ready` throws `SessionNotReady` *before* the distributor is constructed. A premature attach would otherwise start the pump against a source that throws, completing every subscriber with that error and latching the distributor for the session's whole lifetime. +The same one-distributor-per-session rule is enforced at the other end of the lifecycle. Session disposal disposes the distributor and latches the fact, so an attach that passed its readiness gate and then lost the CPU until the session was torn down throws `SessionNotReady` rather than constructing a replacement distributor. A replacement can never be pumped — the pump-start flag is a one-way latch — so its subscriber would block on an empty channel that is never completed until the client's own deadline expired, with no error, log or metric to explain it. Every attach entry point (`AttachEventSubscriber`, `AttachEventSubscriberWithReplay`, `AttachInternalEventSubscriber`) fails this way, and each rolls back the subscriber-count slot it took, so a caller that loses the race to teardown sees the same failure it would have seen a moment later from the readiness gate. + Sessions open with `MxGateway:Sessions:DefaultLeaseSeconds` (default 1800) added to the open timestamp. Unary client activity refreshes the lease by the same duration. `ExtendLease` and `IsLeaseExpired` cooperate with `SessionManager.CloseExpiredLeasesAsync`, which iterates a registry snapshot and closes any session whose lease has expired with `LeaseExpiredReason`. `SessionLeaseMonitorHostedService` runs that sweep every `MxGateway:Sessions:LeaseSweepIntervalSeconds` seconds (default 30). #### Teardown parallelism diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index 5802ff9..5331a81 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -39,6 +39,14 @@ public sealed class GatewaySession private bool _everHadEventSubscriber; private SessionEventDistributor? _eventDistributor; private bool _eventDistributorStarted; + // GWC-34: set under _syncRoot when DisposeAsync nulls _eventDistributor, making the + // "one distributor per session, never reset" lifecycle symmetric at both ends. Without it + // EnsureDistributorCreated reads the null as "not created yet" and builds a REPLACEMENT + // distributor after teardown — one whose pump can never start (_eventDistributorStarted is a + // one-way latch) and whose channel is never completed, so an attach that stalled across the + // close+dispose registers on it and blocks in WaitToReadAsync until the client's own deadline, + // silently and invisibly to metrics. + private bool _eventDistributorDisposed; private bool _dashboardMirrorStarted; private IEventSubscriberLease? _dashboardMirrorLease; private Task? _dashboardMirrorTask; @@ -532,10 +540,23 @@ public sealed class GatewaySession // start). Both the construction and the started-flag flip happen under _syncRoot so two // concurrent callers (e.g. MarkReady's dashboard mirror and a racing first // AttachEventSubscriber) agree on a single distributor and a single start. + // + // GWC-34: after DisposeAsync the null distributor means "gone", not "not built yet", so this + // throws instead of constructing a replacement nothing will ever pump. The attach paths already + // roll back and surface the exception, which is the same SessionNotReady the readiness gates + // raise — a caller that loses the race to teardown now fails immediately instead of hanging. private SessionEventDistributor EnsureDistributorCreated(out bool startNow) { lock (_syncRoot) { + if (_eventDistributorDisposed) + { + throw new SessionManagerException( + SessionManagerErrorCode.SessionNotReady, + $"Session {SessionId} is not ready for event streaming. Its event distributor was " + + "disposed with the session."); + } + if (_eventDistributor is null) { EventOptions eventOptions = _eventStreaming.EventOptions; @@ -672,7 +693,37 @@ public sealed class GatewaySession // dashboard mirror only disconnects itself and never faults the session. // These three calls are OUTSIDE _syncRoot to avoid holding it across // EnsureDistributorCreated's own lock and StartAsync's Task.Run. - SessionEventDistributor distributor = EnsureDistributorCreated(out bool startNow); + // + // GWC-34: the state check above can be overtaken by a close+dispose before we get here, and + // EnsureDistributorCreated now says so instead of building an orphan distributor. The + // dashboard mirror is a best-effort observability attachment on a session that is already + // gone, so it is abandoned here rather than propagated out of MarkReady — the mirror-started + // latch stays set (nothing may start one on a disposed session) and the loop's token source + // is released so it is not orphaned. + SessionEventDistributor distributor; + bool startNow; + try + { + distributor = EnsureDistributorCreated(out startNow); + } + catch (SessionManagerException exception) + { + CancellationTokenSource? abandonedCts; + lock (_syncRoot) + { + abandonedCts = _dashboardMirrorCts; + _dashboardMirrorCts = null; + } + + abandonedCts?.Dispose(); + _eventStreaming.DistributorLogger.LogDebug( + exception, + "Dashboard event mirror for session {SessionId} was abandoned: the session was disposed " + + "before the mirror could attach.", + SessionId); + return; + } + IEventSubscriberLease lease = distributor.Register(isInternal: true); StartPumpIfRequested(distributor, startNow); @@ -1867,6 +1918,9 @@ public sealed class GatewaySession { distributor = _eventDistributor; _eventDistributor = null; + // GWC-34: latch the null-out so EnsureDistributorCreated cannot read it as + // "not created yet" and hand a racing attach a distributor that will never pump. + _eventDistributorDisposed = true; } if (distributor is not null) diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs index 9e78b09..5581103 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs @@ -719,6 +719,47 @@ public sealed class GatewaySessionTests Assert.Equal([1UL, 2UL], received.Select(mxEvent => mxEvent.WorkerSequence).ToArray()); } + /// + /// GWC-34, the disposal-side twin of the GWC-27 premature-attach poison: an attach that + /// passed its readiness gate and then lost the CPU until the session was torn down used to + /// find _eventDistributor null, read that as "not created yet" and build a + /// REPLACEMENT distributor. Its pump could never start (the started flag is a one-way latch) + /// and its channel was never completed, so the subscriber registered on it and blocked in + /// WaitToReadAsync until its own deadline — no error, no log, no metric. Disposal now + /// latches the null-out, so every attach entry point fails immediately and loudly instead. + /// Attaching twice pins the rollback: the second call must still report + /// SessionNotReady, not EventSubscriberAlreadyActive, which it only can if the + /// first failure released the subscriber slot it took. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task AttachAfterDisposeThrowsSessionNotReadyInsteadOfRegisteringOnANeverPumpedDistributor() + { + FakeWorkerClient workerClient = new(); + GatewaySession session = CreateReadySession(workerClient); + + // Disposal leaves the session state Ready (only CloseAsync transitions it), so the attach + // paths' readiness gates pass and the calls below genuinely reach EnsureDistributorCreated — + // the exact position the racing attach in the finding occupies. + await session.DisposeAsync(); + + SessionManagerException attachException = Assert.Throws( + () => session.AttachEventSubscriber(maxSubscribers: 1)); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, attachException.ErrorCode); + + SessionManagerException secondAttachException = Assert.Throws( + () => session.AttachEventSubscriber(maxSubscribers: 1)); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, secondAttachException.ErrorCode); + + SessionManagerException replayException = Assert.Throws( + () => session.AttachEventSubscriberWithReplay(maxSubscribers: 1, afterSequence: 0)); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, replayException.ErrorCode); + + SessionManagerException internalException = Assert.Throws( + () => session.AttachInternalEventSubscriber()); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, internalException.ErrorCode); + } + private static GatewaySession CreateSession() { return new GatewaySession(