diff --git a/docs/Sessions.md b/docs/Sessions.md index 072110b..b77dce5 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -49,7 +49,7 @@ public void TransitionTo(SessionState nextState) ### SessionManager (ISessionManager) -`SessionManager` is the orchestrator. It exposes `OpenSessionAsync`, `TryGetSession`, `InvokeAsync`, `ReadEventsAsync`, `CloseSessionAsync`, `KillWorkerAsync`, `CloseExpiredLeasesAsync`, and `ShutdownAsync`. It composes `ISessionRegistry`, `ISessionWorkerClientFactory`, `GatewayMetrics`, and `GatewayOptions`. +`SessionManager` is the orchestrator. It exposes `OpenSessionAsync`, `TryGetSession`, `InvokeAsync`, `CloseSessionAsync`, `KillWorkerAsync`, `CloseExpiredLeasesAsync`, and `ShutdownAsync`. It composes `ISessionRegistry`, `ISessionWorkerClientFactory`, `GatewayMetrics`, and `GatewayOptions`. `CloseSessionAsync` and `KillWorkerAsync` are both end-of-life paths but differ in what they offer the worker: @@ -191,7 +191,7 @@ The order — fault, deregister, dispose, release slot, record metric, log, reth ### Run -While `Ready`, callers reach the worker through `SessionManager.InvokeAsync` or `ReadEventsAsync`. Both delegate to `GatewaySession`, which checks the state under lock and updates `LastClientActivityAt` on every invocation. `GatewaySession` also exposes typed bulk helpers (`AddItemBulkAsync`, `SubscribeBulkAsync`, etc.) that wrap `WorkerCommand` round-trips and translate non-`Ok` `ProtocolStatus` replies into `SessionManagerException` with `SessionNotReady`. +While `Ready`, callers reach the worker through `SessionManager.InvokeAsync`, which delegates to `GatewaySession`, which checks the state under lock and updates `LastClientActivityAt` on every invocation. Events do not travel this path: every consumer attaches to the session's `SessionEventDistributor` instead (see below), so the manager exposes no event-read member. `GatewaySession` also exposes typed bulk helpers (`AddItemBulkAsync`, `SubscribeBulkAsync`, etc.) that wrap `WorkerCommand` round-trips and translate non-`Ok` `ProtocolStatus` replies into `SessionManagerException` with `SessionNotReady`. Event streaming uses `AttachEventSubscriber` which returns a disposable lease. When `allowMultipleSubscribers` is false (single-subscriber mode) a second attach throws `EventSubscriberAlreadyActive`; this prevents two gRPC streams from racing on the same worker event channel. When it is true, up to `MaxEventSubscribersPerSession` concurrent external subscribers are allowed and the next attach throws `EventSubscriberLimitReached`. The count-check-and-increment is atomic under the session lock, so concurrent attaches can never exceed the cap. The gateway-owned internal dashboard mirror subscriber is registered directly on the distributor and does not count toward the cap. Active event subscribers keep the session lease from expiring until the stream is disposed. diff --git a/docs/plans/2026-08-15-deferred-remediation.md b/docs/plans/2026-08-15-deferred-remediation.md index fb42808..a22e610 100644 --- a/docs/plans/2026-08-15-deferred-remediation.md +++ b/docs/plans/2026-08-15-deferred-remediation.md @@ -321,7 +321,8 @@ is worth keeping, this is the record. through `ISessionManager.ReadEventsAsync`. That interface member itself has zero production call sites — only test fakes implement and exercise it. Deleting it is a mechanical but wide change (~15 test-fake touches), so it is recorded as a follow-up -rather than done here. +rather than done here. Removed by `docs/plans/2026-08-17-deferred-closeout.md` Task 1, +2026-08-17. **Task 5 — dashboard event feed, two review rounds.** Review caught two races that the first cut did not have. First, subscription lifetime: subscriptions are now diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index ac426a4..f7c8808 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -765,13 +765,11 @@ public sealed class GatewaySession // distributor guarantees a single consumer) and maps each frame to the public MxEvent, // preserving worker order. Mirrors the former ProduceEventsAsync mapping exactly. // - // This deliberately duplicates the three lines of ReadEventsAsync rather than enumerating - // it: every worker event crosses this source, and routing it through a second pure - // pass-through iterator cost two extra MoveNextAsync state-machine hops per event for no - // semantic value. ReadEventsAsync stays for ISessionManager.ReadEventsAsync; keep the two - // bodies in step. Only one of them may run per attach — WorkerClient.ReadEventsAsync - // single-reader-claims the event channel and throws on a second consumer — and on the - // distributor path that one consumer is this method. + // This is the session's only reader of the worker event channel: every gateway consumer — + // gRPC subscribers, the dashboard mirror, the alarm monitor — attaches to the distributor + // this source feeds. WorkerClient.ReadEventsAsync single-reader-claims that channel and + // throws on a second consumer, so any future path that drains it directly fails loudly + // rather than splitting events. private async IAsyncEnumerable MapWorkerEventsAsync( [EnumeratorCancellation] CancellationToken cancellationToken) { @@ -1522,33 +1520,6 @@ public sealed class GatewaySession cancellationToken); } - /// - /// Reads events from the worker as an asynchronous enumerable stream. - /// - /// Token to cancel the asynchronous operation. - /// - /// Backs ISessionManager.ReadEventsAsync. The distributor does not come - /// through here — MapWorkerEventsAsync inlines this body to save a per-event - /// iterator hop, so changes made here belong there too. The two are mutually exclusive - /// per attach: claims the worker event - /// channel for a single reader and throws on the second consumer. - /// - /// An asynchronous stream of worker events. - public async IAsyncEnumerable ReadEventsAsync( - [EnumeratorCancellation] CancellationToken cancellationToken) - { - IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false); - TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow()); - - await foreach (WorkerEvent workerEvent in workerClient - .ReadEventsAsync(cancellationToken) - .WithCancellation(cancellationToken) - .ConfigureAwait(false)) - { - yield return workerEvent; - } - } - /// /// Closes the session and shuts down the worker process. /// diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs index 7a9b749..9cb0191 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs @@ -35,14 +35,6 @@ public interface ISessionManager WorkerCommand command, CancellationToken cancellationToken); - /// Reads events streamed from the worker for the specified session. - /// Identifier of the session. - /// Token to cancel the asynchronous operation. - /// Events emitted by the worker. - IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken); - /// Closes a session and terminates its worker process. /// Identifier of the session to close. /// Token to cancel the asynchronous operation. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs index e369141..0cf27b6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs @@ -187,16 +187,6 @@ public sealed class SessionManager : ISessionManager } } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) - { - GatewaySession session = GetRequiredSession(sessionId); - - return session.ReadEventsAsync(cancellationToken); - } - /// public async Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs index dd0813e..bf9a0d0 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs @@ -464,11 +464,6 @@ public sealed class AlarmFailoverEndToEndTests return Task.FromResult(new WorkerCommandReply { Reply = reply }); } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) => throw new NotSupportedException(); - /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs index 969645b..956cf14 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorAttachOrderTests.cs @@ -618,11 +618,6 @@ public sealed class GatewayAlarmMonitorAttachOrderTests return new WorkerCommandReply { Reply = reply }; } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) => throw new NotSupportedException(); - /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs index eef1fa3..3b3f8fb 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs @@ -777,11 +777,6 @@ public sealed class GatewayAlarmMonitorProviderModeTests return Task.FromResult(new WorkerCommandReply { Reply = reply }); } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) => throw new NotSupportedException(); - /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs index 3059458..d03b4d9 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs @@ -261,10 +261,6 @@ public sealed class DashboardLiveDataServiceTests CancellationToken cancellationToken) => throw new NotSupportedException(); - /// - public IAsyncEnumerable ReadEventsAsync(string sessionId, CancellationToken cancellationToken) => - throw new NotSupportedException(); - /// public Task KillWorkerAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs index b0cdf77..2482c1a 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs @@ -370,14 +370,6 @@ public sealed class DashboardSessionAdminServiceTests throw new NotSupportedException(); } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) - { - throw new NotSupportedException(); - } - /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs index 5695f44..177e6e9 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs @@ -761,14 +761,6 @@ public sealed class EventStreamServiceTests return Task.FromResult(new WorkerCommandReply()); } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) - { - return _sessions[sessionId].ReadEventsAsync(cancellationToken); - } - /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs index 885bffb..304a9a0 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs @@ -935,19 +935,6 @@ public sealed class MxAccessGatewayServiceConstraintTests return Task.FromResult(InvokeReply); } - /// - public async IAsyncEnumerable ReadEventsAsync( - string sessionId, - [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) - { - foreach (WorkerEvent ev in Events) - { - cancellationToken.ThrowIfCancellationRequested(); - await Task.Yield(); - yield return ev; - } - } - /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs index 5ad05a8..cfc06a3 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs @@ -517,7 +517,7 @@ public sealed class MxAccessGatewayServiceTests /// The last owner key id passed to OpenSessionAsync. public string? LastOwnerKeyId { get; private set; } - /// The last session ID passed to ReadEventsAsync. + /// The last session ID the event stream service was asked to stream. public string? LastReadEventsSessionId { get; private set; } /// The last worker command passed to InvokeAsync. @@ -540,10 +540,10 @@ public sealed class MxAccessGatewayServiceTests /// The number of times InvokeAsync was called. public int InvokeCount { get; private set; } - /// The events to return from ReadEventsAsync. + /// The events the fake event stream service replays for this manager. public List Events { get; } = []; - /// Records the session ID passed to ReadEventsAsync. + /// Records the session ID the event stream service was asked to stream. /// Identifier of the session. public void RecordReadEventsSessionId(string sessionId) { @@ -602,20 +602,6 @@ public sealed class MxAccessGatewayServiceTests return Task.FromResult(InvokeReply); } - /// - public async IAsyncEnumerable ReadEventsAsync( - string sessionId, - [EnumeratorCancellation] CancellationToken cancellationToken) - { - LastReadEventsSessionId = sessionId; - foreach (WorkerEvent workerEvent in Events) - { - cancellationToken.ThrowIfCancellationRequested(); - await Task.Yield(); - yield return workerEvent; - } - } - /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs index b103a29..1d54933 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs @@ -347,11 +347,6 @@ public sealed class GatewaySessionDashboardMirrorTests WorkerCommand command, CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply()); - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) => session.ReadEventsAsync(cancellationToken); - /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs index 1808e54..b9327a6 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs @@ -848,14 +848,6 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests return Task.FromResult(new WorkerCommandReply()); } - /// - public IAsyncEnumerable ReadEventsAsync( - string sessionId, - CancellationToken cancellationToken) - { - return AsyncEnumerable.Empty(); - } - /// public Task CloseSessionAsync( string sessionId,