fix(dashboard): guard stale-session batches inside the renderer dispatch; register IDashboardSessionEventSubscriber
This commit is contained in:
+61
-20
@@ -5,7 +5,7 @@
|
|||||||
@using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs
|
@using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs
|
||||||
@inject AuthenticationStateProvider AuthenticationStateProvider
|
@inject AuthenticationStateProvider AuthenticationStateProvider
|
||||||
@inject IDashboardSessionAdminService SessionAdminService
|
@inject IDashboardSessionAdminService SessionAdminService
|
||||||
@inject IDashboardEventBroadcaster EventBroadcaster
|
@inject IDashboardSessionEventSubscriber EventSubscriber
|
||||||
|
|
||||||
<PageTitle>Dashboard Session</PageTitle>
|
<PageTitle>Dashboard Session</PageTitle>
|
||||||
|
|
||||||
@@ -157,8 +157,18 @@ else
|
|||||||
private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session =>
|
private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session =>
|
||||||
string.Equals(session.SessionId, SessionId, StringComparison.Ordinal));
|
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 IDashboardEventSubscription? _eventSubscription;
|
||||||
private CancellationTokenSource? _eventPumpCancellation;
|
private CancellationTokenSource? _eventPumpCancellation;
|
||||||
|
private Task? _eventPumpTask;
|
||||||
private bool _eventsConnected;
|
private bool _eventsConnected;
|
||||||
private string? _subscribedSessionId;
|
private string? _subscribedSessionId;
|
||||||
private readonly LinkedList<MxEvent> _recentEvents = new();
|
private readonly LinkedList<MxEvent> _recentEvents = new();
|
||||||
@@ -180,17 +190,16 @@ else
|
|||||||
CanManage = SessionAdminService.CanManage(authenticationState.User);
|
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))
|
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();
|
AttachEvents();
|
||||||
}
|
}
|
||||||
|
|
||||||
return Task.CompletedTask;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private PendingConfirm? PendingAction { get; set; }
|
private PendingConfirm? PendingAction { get; set; }
|
||||||
@@ -266,27 +275,31 @@ else
|
|||||||
string ConfirmButtonClass,
|
string ConfirmButtonClass,
|
||||||
Func<System.Security.Claims.ClaimsPrincipal, Task<DashboardSessionAdminResult>> Action);
|
Func<System.Security.Claims.ClaimsPrincipal, Task<DashboardSessionAdminResult>> 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
|
// 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
|
// SignalR connection to /hubs/events — mint a hub token, negotiate, hold a
|
||||||
// WebSocket, serialize every event — to reach data already sitting in memory.
|
// WebSocket, serialize every event — to reach data already sitting in memory.
|
||||||
// The subscription still registers with EventsHubViewerRegistry, so the
|
// IDashboardSessionEventSubscriber resolves to the same singleton that serves
|
||||||
// broadcaster's "nobody is watching" gate keeps working for both audiences.
|
// 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
|
// 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).
|
// any session (SEC-25 tracks the per-session ACL for both seams).
|
||||||
private void AttachEvents()
|
private void AttachEvents()
|
||||||
{
|
{
|
||||||
if (string.IsNullOrWhiteSpace(SessionId) || EventBroadcaster is not IDashboardSessionEventSubscriber subscriber)
|
if (string.IsNullOrWhiteSpace(SessionId))
|
||||||
{
|
{
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
_eventSubscription = subscriber.Subscribe(SessionId);
|
_eventSubscription = EventSubscriber.Subscribe(SessionId);
|
||||||
_eventPumpCancellation = new CancellationTokenSource();
|
_eventPumpCancellation = new CancellationTokenSource();
|
||||||
_eventsConnected = true;
|
_eventsConnected = true;
|
||||||
_subscribedSessionId = SessionId;
|
_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)
|
private async Task PumpEventsAsync(IDashboardEventSubscription subscription, CancellationToken cancellationToken)
|
||||||
@@ -311,6 +324,17 @@ else
|
|||||||
|
|
||||||
await InvokeAsync(() =>
|
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)
|
foreach (MxEvent mxEvent in batch)
|
||||||
{
|
{
|
||||||
_recentEvents.AddFirst(mxEvent);
|
_recentEvents.AddFirst(mxEvent);
|
||||||
@@ -331,26 +355,43 @@ else
|
|||||||
}
|
}
|
||||||
catch (ObjectDisposedException)
|
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;
|
IDashboardEventSubscription? subscription = _eventSubscription;
|
||||||
CancellationTokenSource? cancellation = _eventPumpCancellation;
|
CancellationTokenSource? cancellation = _eventPumpCancellation;
|
||||||
|
Task? pump = _eventPumpTask;
|
||||||
_eventSubscription = null;
|
_eventSubscription = null;
|
||||||
_eventPumpCancellation = null;
|
_eventPumpCancellation = null;
|
||||||
|
_eventPumpTask = null;
|
||||||
_eventsConnected = false;
|
_eventsConnected = false;
|
||||||
_subscribedSessionId = null;
|
_subscribedSessionId = null;
|
||||||
_recentEvents.Clear();
|
_recentEvents.Clear();
|
||||||
|
|
||||||
// Cancel first so the pump stops touching the renderer, then dispose the
|
// Cancel and drop the subscription before draining. Disposing it releases the
|
||||||
// subscription — that is what releases the viewer registration and lets the
|
// viewer registration — the whole point of the gate — and completes the channel,
|
||||||
// broadcaster go back to skipping mirror work for this session.
|
// so the pump has an exit even if cancellation is missed.
|
||||||
cancellation?.Cancel();
|
cancellation?.Cancel();
|
||||||
cancellation?.Dispose();
|
|
||||||
subscription?.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)
|
private static string EventStatusLabel(MxEvent evt)
|
||||||
@@ -362,7 +403,7 @@ else
|
|||||||
|
|
||||||
public new async ValueTask DisposeAsync()
|
public new async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
DetachEvents();
|
await DetachEventsAsync();
|
||||||
await base.DisposeAsync().ConfigureAwait(false);
|
await base.DisposeAsync().ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -51,7 +51,17 @@ public static class DashboardServiceCollectionExtensions
|
|||||||
// Singleton: EventsHub instances are transient (one per hub invocation), so the
|
// Singleton: EventsHub instances are transient (one per hub invocation), so the
|
||||||
// subscriber bookkeeping they share with the broadcaster must outlive them.
|
// subscriber bookkeeping they share with the broadcaster must outlive them.
|
||||||
services.AddSingleton<Hubs.EventsHubViewerRegistry>();
|
services.AddSingleton<Hubs.EventsHubViewerRegistry>();
|
||||||
services.AddSingleton<Hubs.IDashboardEventBroadcaster, Hubs.DashboardEventBroadcaster>();
|
|
||||||
|
// 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<Hubs.DashboardEventBroadcaster>();
|
||||||
|
services.AddSingleton<Hubs.IDashboardEventBroadcaster>(
|
||||||
|
static provider => provider.GetRequiredService<Hubs.DashboardEventBroadcaster>());
|
||||||
|
services.AddSingleton<Hubs.IDashboardSessionEventSubscriber>(
|
||||||
|
static provider => provider.GetRequiredService<Hubs.DashboardEventBroadcaster>());
|
||||||
services.AddSingleton<Hubs.DashboardSnapshotHubConnectionCounter>();
|
services.AddSingleton<Hubs.DashboardSnapshotHubConnectionCounter>();
|
||||||
services.AddHostedService<Hubs.DashboardSnapshotPublisher>();
|
services.AddHostedService<Hubs.DashboardSnapshotPublisher>();
|
||||||
services.AddHostedService<Hubs.AlarmsHubPublisher>();
|
services.AddHostedService<Hubs.AlarmsHubPublisher>();
|
||||||
|
|||||||
@@ -45,4 +45,23 @@ public sealed class DashboardHubsRegistrationTests
|
|||||||
.GetRequiredService<DashboardHubConnectionFactory>();
|
.GetRequiredService<DashboardHubConnectionFactory>();
|
||||||
Assert.NotNull(factory);
|
Assert.NotNull(factory);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Build_WhenDashboardEnabled_ResolvesBothEventMirrorInterfacesToOneInstance()
|
||||||
|
{
|
||||||
|
await using WebApplication app = GatewayApplication.Build([]);
|
||||||
|
|
||||||
|
IDashboardEventBroadcaster broadcaster = app.Services.GetRequiredService<IDashboardEventBroadcaster>();
|
||||||
|
IDashboardSessionEventSubscriber subscriber = app.Services
|
||||||
|
.GetRequiredService<IDashboardSessionEventSubscriber>();
|
||||||
|
|
||||||
|
Assert.Same(broadcaster, subscriber);
|
||||||
|
Assert.Same(app.Services.GetRequiredService<DashboardEventBroadcaster>(), broadcaster);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user