diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor index af50f53..aebe8f4 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor @@ -5,7 +5,7 @@ @using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs @inject AuthenticationStateProvider AuthenticationStateProvider @inject IDashboardSessionAdminService SessionAdminService -@inject IDashboardEventBroadcaster EventBroadcaster +@inject IDashboardSessionEventSubscriber EventSubscriber Dashboard Session @@ -157,8 +157,18 @@ else private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session => string.Equals(session.SessionId, SessionId, StringComparison.Ordinal)); + // Upper bound on waiting for the event pump while detaching, mirroring + // DashboardPageBase's snapshot-watch drain: the pump marshals renders through the + // renderer's dispatcher and a detach can run on that same dispatcher, so the wait + // is bounded rather than unconditional. + private static readonly TimeSpan EventPumpDrainTimeout = TimeSpan.FromSeconds(5); + + // Written only on the renderer's dispatcher (the lifecycle methods below), and read + // on it from inside the pump's dispatched callback — that pairing is what makes the + // stale-batch guard in PumpEventsAsync reliable. private IDashboardEventSubscription? _eventSubscription; private CancellationTokenSource? _eventPumpCancellation; + private Task? _eventPumpTask; private bool _eventsConnected; private string? _subscribedSessionId; private readonly LinkedList _recentEvents = new(); @@ -180,17 +190,16 @@ else CanManage = SessionAdminService.CanManage(authenticationState.User); } - protected override Task OnParametersSetAsync() + protected override async Task OnParametersSetAsync() { - // Attach/detach are synchronous now that the feed is in-process; the override - // stays on the async lifecycle member so the base class's contract is untouched. if (!string.Equals(_subscribedSessionId, SessionId, StringComparison.Ordinal)) { - DetachEvents(); + // Deliberately no ConfigureAwait(false): the resumption must stay on the + // renderer's dispatcher so the new subscription is published to + // _eventSubscription from the same thread the pump's guard reads it on. + await DetachEventsAsync(); AttachEvents(); } - - return Task.CompletedTask; } private PendingConfirm? PendingAction { get; set; } @@ -266,27 +275,31 @@ else string ConfirmButtonClass, Func> Action); - // The dashboard runs in the same process as the broadcaster, so this page reads + // The dashboard runs in the same process as the event mirror, so this page reads // the session's mirrored events straight from it. It used to open a loopback // SignalR connection to /hubs/events — mint a hub token, negotiate, hold a // WebSocket, serialize every event — to reach data already sitting in memory. - // The subscription still registers with EventsHubViewerRegistry, so the - // broadcaster's "nobody is watching" gate keeps working for both audiences. + // IDashboardSessionEventSubscriber resolves to the same singleton that serves + // IDashboardEventBroadcaster, and the subscription registers with + // EventsHubViewerRegistry, so the "nobody is watching" gate keeps working for + // both audiences. // ACL posture is unchanged from the hub path: any dashboard Viewer may watch // any session (SEC-25 tracks the per-session ACL for both seams). private void AttachEvents() { - if (string.IsNullOrWhiteSpace(SessionId) || EventBroadcaster is not IDashboardSessionEventSubscriber subscriber) + if (string.IsNullOrWhiteSpace(SessionId)) { return; } - _eventSubscription = subscriber.Subscribe(SessionId); + _eventSubscription = EventSubscriber.Subscribe(SessionId); _eventPumpCancellation = new CancellationTokenSource(); _eventsConnected = true; _subscribedSessionId = SessionId; - _ = PumpEventsAsync(_eventSubscription, _eventPumpCancellation.Token); + // Deliberately not awaited: the pump runs for as long as the page watches this + // session and is cancelled and drained by DetachEventsAsync. + _eventPumpTask = PumpEventsAsync(_eventSubscription, _eventPumpCancellation.Token); } private async Task PumpEventsAsync(IDashboardEventSubscription subscription, CancellationToken cancellationToken) @@ -311,6 +324,17 @@ else await InvokeAsync(() => { + // The batch was read before this callback was dispatched, and a + // session switch can land in between. Rendering it then would show + // the previous session's events under the new session's heading, so + // a batch whose subscription is no longer the live one is dropped. + // Safe as an unsynchronized read: _eventSubscription is written on + // this same dispatcher. + if (!ReferenceEquals(_eventSubscription, subscription)) + { + return; + } + foreach (MxEvent mxEvent in batch) { _recentEvents.AddFirst(mxEvent); @@ -331,26 +355,43 @@ else } catch (ObjectDisposedException) { - // The renderer went away while a batch was being dispatched. + // Either the renderer went away mid-dispatch, or the drain below timed out + // and disposed the cancellation source this loop is still reading. } } - private void DetachEvents() + private async Task DetachEventsAsync() { IDashboardEventSubscription? subscription = _eventSubscription; CancellationTokenSource? cancellation = _eventPumpCancellation; + Task? pump = _eventPumpTask; _eventSubscription = null; _eventPumpCancellation = null; + _eventPumpTask = null; _eventsConnected = false; _subscribedSessionId = null; _recentEvents.Clear(); - // Cancel first so the pump stops touching the renderer, then dispose the - // subscription — that is what releases the viewer registration and lets the - // broadcaster go back to skipping mirror work for this session. + // Cancel and drop the subscription before draining. Disposing it releases the + // viewer registration — the whole point of the gate — and completes the channel, + // so the pump has an exit even if cancellation is missed. cancellation?.Cancel(); - cancellation?.Dispose(); subscription?.Dispose(); + + try + { + if (pump is not null) + { + await pump.WaitAsync(EventPumpDrainTimeout); + } + } + catch + { + // Detach-time errors (including a drain timeout) are best-effort. + } + + // Disposed after the drain so the pump is no longer reading the token. + cancellation?.Dispose(); } private static string EventStatusLabel(MxEvent evt) @@ -362,7 +403,7 @@ else public new async ValueTask DisposeAsync() { - DetachEvents(); + await DetachEventsAsync(); await base.DisposeAsync().ConfigureAwait(false); } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs index aece649..e80508b 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs @@ -51,7 +51,17 @@ public static class DashboardServiceCollectionExtensions // Singleton: EventsHub instances are transient (one per hub invocation), so the // subscriber bookkeeping they share with the broadcaster must outlive them. services.AddSingleton(); - services.AddSingleton(); + + // One instance behind two interfaces, registered concretely and forwarded: the + // publish side (IDashboardEventBroadcaster, driven by the session pipeline) and + // the in-process subscribe side (IDashboardSessionEventSubscriber, used by the + // session-details page) share subscriber bookkeeping, so resolving them to two + // instances would leave the page subscribed to a mirror nobody publishes to. + services.AddSingleton(); + services.AddSingleton( + static provider => provider.GetRequiredService()); + services.AddSingleton( + static provider => provider.GetRequiredService()); services.AddSingleton(); services.AddHostedService(); services.AddHostedService(); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardHubsRegistrationTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardHubsRegistrationTests.cs index 638dbd5..f6af323 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardHubsRegistrationTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardHubsRegistrationTests.cs @@ -45,4 +45,23 @@ public sealed class DashboardHubsRegistrationTests .GetRequiredService(); Assert.NotNull(factory); } + + /// + /// The publish and in-process subscribe faces of the event mirror must resolve to + /// one instance: two would leave the session-details page reading a mirror the + /// session pipeline never publishes to, and the viewer gate would never open. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Build_WhenDashboardEnabled_ResolvesBothEventMirrorInterfacesToOneInstance() + { + await using WebApplication app = GatewayApplication.Build([]); + + IDashboardEventBroadcaster broadcaster = app.Services.GetRequiredService(); + IDashboardSessionEventSubscriber subscriber = app.Services + .GetRequiredService(); + + Assert.Same(broadcaster, subscriber); + Assert.Same(app.Services.GetRequiredService(), broadcaster); + } }