diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index ab1e41e..0b9fa38 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -548,10 +548,37 @@ public sealed class GatewaySession /// MaxEventSubscribersPerSession accounting and out of the single-subscriber /// overflow-fault path, so a slow alarm reconcile can never fault the session — it only /// disconnects this internal subscriber. + /// + /// Gated on readiness exactly like : attaching + /// before the session and its worker are Ready throws + /// with + /// . + /// /// /// The internal subscriber's lease; dispose it to unregister. + /// + /// The session or its worker client is not Ready. + /// public IEventSubscriberLease AttachInternalEventSubscriber() { + // Readiness gate, mirroring AttachEventSubscriber (GWC-27). It must run BEFORE + // EnsureDistributorCreated: a premature attach would construct the distributor and start + // its pump against a not-yet-Ready worker, the pump source would throw SessionNotReady, + // PumpAsync would complete every subscriber with that error and latch _completed, and + // _eventDistributorStarted is never reset — so the session would reach Ready with + // permanently dead event streaming, silently, for the rest of its life. Failing loudly + // here keeps that state unreachable. The check is under _syncRoot and the distributor + // calls stay outside it, matching AttachEventSubscriber's lock discipline. + lock (_syncRoot) + { + if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready) + { + throw new SessionManagerException( + SessionManagerErrorCode.SessionNotReady, + $"Session {SessionId} is not ready for event streaming. Current state is {_state}."); + } + } + // Same sequence StartDashboardMirror uses: create the distributor (claiming the pump // start if we are first), register the internal subscriber BEFORE the pump starts so a // subscriber is always present at pump start, then start the pump if requested. 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 7dfaef8..9e78b09 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs @@ -668,6 +668,81 @@ public sealed class GatewaySessionTests Assert.Equal(SessionState.Ready, session.State); } + /// + /// GWC-27: must refuse to + /// attach before the session is Ready. Without the gate the attach would construct and + /// start the distributor against a not-yet-Ready worker; the pump source throws + /// SessionNotReady, every subscriber is completed with that error, and the + /// distributor latches — leaving a session that reaches Ready with permanently dead + /// event streaming. The second half of the test is the load-bearing one: after the + /// failed attach the session still streams live events, proving the distributor was + /// never created or started. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor() + { + FakeWorkerClient workerClient = new(); + workerClient.Events.Add(new WorkerEvent + { + Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 1, OnDataChange = new OnDataChangeEvent() }, + }); + workerClient.Events.Add(new WorkerEvent + { + Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 2, OnDataChange = new OnDataChangeEvent() }, + }); + + // Constructed but neither worker-attached nor Ready — the premature-attach case. + await using GatewaySession session = CreateSession(); + + SessionManagerException exception = Assert.Throws( + () => session.AttachInternalEventSubscriber()); + Assert.Equal(SessionManagerErrorCode.SessionNotReady, exception.ErrorCode); + + // Drive the session to Ready and stream: the failed attach must not have poisoned + // (or even created) the distributor, so a normal subscriber still receives events. + session.AttachWorkerClient(workerClient); + session.MarkReady(); + + using IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1); + List received = []; + using CancellationTokenSource readCts = new(TimeSpan.FromSeconds(5)); + await foreach (MxEvent mxEvent in lease.Reader.ReadAllAsync(readCts.Token)) + { + received.Add(mxEvent); + if (received.Count == 2) + { + break; + } + } + + Assert.Equal([1UL, 2UL], received.Select(mxEvent => mxEvent.WorkerSequence).ToArray()); + } + + private static GatewaySession CreateSession() + { + return new GatewaySession( + sessionId: "session-test-internal-attach", + backendName: "mxaccess", + pipeName: "mxaccess-gateway-1-session-test-internal-attach", + 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.Instance, + TimeProvider.System, + new GatewayMetrics())); + } + private static GatewaySession CreateReadySessionWithDetachGrace( IWorkerClient workerClient, TimeProvider timeProvider, @@ -855,6 +930,9 @@ public sealed class GatewaySessionTests /// Gets the count of dispose invocations. public int DisposeCount { get; private set; } + /// Events yields, in order, before completing. Empty by default. + public List Events { get; } = []; + /// public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; @@ -869,7 +947,11 @@ public sealed class GatewaySessionTests [EnumeratorCancellation] CancellationToken cancellationToken) { await Task.CompletedTask.ConfigureAwait(false); - yield break; + foreach (WorkerEvent workerEvent in Events) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return workerEvent; + } } ///