From 3688fa19e6d7f502e43885178b9a01859ff86779 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:14:53 -0400 Subject: [PATCH] fix(dashboard): time-bound alarm hub sends and mark each resubscribe AlarmsHubPublisher awaited every group send unbounded inside the StreamAsync loop, so one slow client stalled the drain until the monitor dropped the subscriber and the resubscribe replayed the full snapshot to everyone. Each broadcast is now bounded (2s) and abandoned on timeout, and every (re)subscription is preceded by an AlarmFeedReset frame so clients discard state before the replayed snapshot. --- .../Dashboard/Hubs/AlarmFeedReset.cs | 14 ++ .../Dashboard/Hubs/AlarmsHub.cs | 8 + .../Dashboard/Hubs/AlarmsHubPublisher.cs | 132 ++++++++++++--- .../Dashboard/AlarmsHubPublisherTests.cs | 159 +++++++++++++++--- 4 files changed, 270 insertions(+), 43 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmFeedReset.cs diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmFeedReset.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmFeedReset.cs new file mode 100644 index 0000000..e1742f4 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmFeedReset.cs @@ -0,0 +1,14 @@ +namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +/// +/// Payload of the frame: the publisher has +/// (re)subscribed to the central alarm monitor, which primes the subscription with +/// a full active-alarm snapshot. A client discards its accumulated alarm state on +/// receipt and rebuilds it from the active_alarm rows up to the next +/// snapshot_complete. +/// +/// +/// Monotonic subscription counter for this gateway process, starting at 1. Lets a +/// client tell a resubscribe from a duplicate delivery. +/// +public sealed record AlarmFeedReset(long Generation); diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHub.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHub.cs index d48f997..d0c6287 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHub.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHub.cs @@ -23,11 +23,19 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; [Authorize(Policy = DashboardAuthenticationDefaults.AlarmsPolicy)] public sealed class AlarmsHub : Hub { + /// Group every admitted connection joins on connect. public const string AllAlarmsGroup = "__alarms__"; /// Method name used to push AlarmFeedMessage values to clients. public const string AlarmMessage = "AlarmFeed"; + /// + /// Method name of the marker sent ahead of every + /// (re)subscription to the central monitor. A client discards its accumulated + /// alarm state on receipt and rebuilds it from the snapshot that follows. + /// + public const string ResetMessage = "AlarmFeedReset"; + /// public override async Task OnConnectedAsync() { diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHubPublisher.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHubPublisher.cs index c280d73..adce9ca 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHubPublisher.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/AlarmsHubPublisher.cs @@ -31,18 +31,69 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// narrowing this loop could apply even in principle. Adding a filter here would imply a /// per-caller scope the feed does not carry. /// +/// +/// Each broadcast is time-bounded (). A group send +/// completes only once every member accepted the frame, so awaiting it unbounded +/// let one slow browser stall this loop; the stalled loop stopped draining its +/// monitor subscription, the monitor dropped it as a slow subscriber, and the +/// resubscribe replayed the whole active-alarm snapshot to every client. A send that +/// misses the bound is abandoned and the loop moves on. +/// +/// +/// Every subscription — the first and each resubscribe — is preceded by an +/// frame. The monitor primes every new +/// subscription with a full snapshot, so a client that keeps accumulating +/// active_alarm rows across a resubscribe would double-count or keep alarms +/// that cleared while the feed was down; the marker tells it to discard its state +/// and rebuild from the snapshot that follows. +/// /// -/// The gateway's central alarm feed. -/// Hub context used to broadcast to the alarms group. -/// Gateway options supplying Dashboard:ShowTagValues. -/// Logger for best-effort broadcast failures. -public sealed class AlarmsHubPublisher( - IGatewayAlarmService alarmService, - IHubContext hubContext, - IOptions options, - ILogger logger) : BackgroundService +public sealed class AlarmsHubPublisher : BackgroundService { - private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues; + /// Upper bound on one alarm broadcast before it is abandoned. + internal static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(2); + + private readonly IGatewayAlarmService _alarmService; + private readonly IHubContext _hubContext; + private readonly ILogger _logger; + private readonly TimeSpan _sendTimeout; + private readonly bool _showTagValues; + private long _subscriptionGeneration; + + /// Initializes a new instance of the class. + /// The gateway's central alarm feed. + /// Hub context used to broadcast to the alarms group. + /// Gateway options supplying Dashboard:ShowTagValues. + /// Logger for best-effort broadcast failures. + public AlarmsHubPublisher( + IGatewayAlarmService alarmService, + IHubContext hubContext, + IOptions options, + ILogger logger) + : this(alarmService, hubContext, options, logger, DefaultSendTimeout) + { + } + + /// Initializes a new instance with a custom send timeout (test hook). + /// The gateway's central alarm feed. + /// Hub context used to broadcast to the alarms group. + /// Gateway options supplying Dashboard:ShowTagValues. + /// Logger for best-effort broadcast failures. + /// Upper bound on one broadcast. + internal AlarmsHubPublisher( + IGatewayAlarmService alarmService, + IHubContext hubContext, + IOptions options, + ILogger logger, + TimeSpan sendTimeout) + { + ArgumentNullException.ThrowIfNull(options); + _alarmService = alarmService ?? throw new ArgumentNullException(nameof(alarmService)); + _hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _sendTimeout = sendTimeout; + _showTagValues = options.Value.Dashboard.ShowTagValues; + } /// protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -54,7 +105,16 @@ public sealed class AlarmsHubPublisher( { try { - await foreach (AlarmFeedMessage message in alarmService + // Sent before the subscription opens, so it reaches clients ahead of + // the snapshot the monitor primes the new subscription with. + long generation = Interlocked.Increment(ref _subscriptionGeneration); + await BroadcastAsync( + AlarmsHub.ResetMessage, + new AlarmFeedReset(generation), + stoppingToken) + .ConfigureAwait(false); + + await foreach (AlarmFeedMessage message in _alarmService .StreamAsync(alarmFilterPrefix: null, stoppingToken) .ConfigureAwait(false)) { @@ -63,18 +123,8 @@ public sealed class AlarmsHubPublisher( break; } - try - { - AlarmFeedMessage outbound = _showTagValues ? message : RedactValues(message); - await hubContext.Clients - .Group(AlarmsHub.AllAlarmsGroup) - .SendAsync(AlarmsHub.AlarmMessage, outbound, stoppingToken) - .ConfigureAwait(false); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - logger.LogWarning(ex, "Alarm broadcast failed; continuing."); - } + AlarmFeedMessage outbound = _showTagValues ? message : RedactValues(message); + await BroadcastAsync(AlarmsHub.AlarmMessage, outbound, stoppingToken).ConfigureAwait(false); } } catch (OperationCanceledException) @@ -83,7 +133,7 @@ public sealed class AlarmsHubPublisher( } catch (Exception ex) { - logger.LogWarning(ex, "Alarm subscription faulted; reconnecting in 5s."); + _logger.LogWarning(ex, "Alarm subscription faulted; reconnecting in 5s."); try { await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken).ConfigureAwait(false); @@ -96,6 +146,40 @@ public sealed class AlarmsHubPublisher( } } + /// + /// Sends one frame to the alarms group within the send timeout. A slow or failed + /// send is logged and dropped so the feed keeps draining; only host shutdown + /// propagates. + /// + /// Client method name. + /// The frame payload. + /// Host shutdown token. + /// A task that completes when the send finished or was abandoned. + private async Task BroadcastAsync(string method, object argument, CancellationToken stoppingToken) + { + try + { + bool sent = await BoundedHubSend.SendAsync( + _hubContext.Clients.Group(AlarmsHub.AllAlarmsGroup), + method, + argument, + _sendTimeout, + stoppingToken) + .ConfigureAwait(false); + if (!sent) + { + _logger.LogDebug( + "Alarm broadcast {Method} exceeded {Timeout} and was abandoned; continuing.", + method, + _sendTimeout); + } + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Alarm broadcast failed; continuing."); + } + } + /// /// Produces a copy of with the alarm value fields /// cleared, leaving every other field — reference, severity, state, category, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/AlarmsHubPublisherTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/AlarmsHubPublisherTests.cs index ea9e9fb..21711f8 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/AlarmsHubPublisherTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/AlarmsHubPublisherTests.cs @@ -15,7 +15,9 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard; /// both value-bearing payload arms are stripped from the copy broadcast to /// browser clients when the flag is off, present when it is on, and the source /// — shared with the gRPC StreamAlarms -/// subscribers and the alarms page — is never mutated. +/// subscribers and the alarms page — is never mutated. Also verifies that a +/// slow client cannot stall the feed and that every (re)subscription is announced +/// with a reset marker ahead of its snapshot. /// public sealed class AlarmsHubPublisherTests { @@ -136,6 +138,81 @@ public sealed class AlarmsHubPublisherTests Assert.True(hubContext.Sent[2].SnapshotStatus.Truncated); } + /// + /// Every subscription — the first and each resubscribe after the monitor ends a + /// stream — is preceded by a reset marker, so a client knows to discard its + /// state before the replayed snapshot arrives instead of double-counting it. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ExecuteAsync_OnResubscribe_SendsResetMarkerBeforeTheReplayedSnapshot() + { + AlarmFeedMessage first = BuildActiveAlarm(); + AlarmFeedMessage replayed = BuildActiveAlarm(); + ScriptedAlarmService alarmService = new([first], [replayed]); + CapturingHubContext hubContext = new(); + AlarmsHubPublisher publisher = CreatePublisher(alarmService, hubContext, AlarmsHubPublisher.DefaultSendTimeout); + + using CancellationTokenSource cts = new(); + await publisher.StartAsync(cts.Token).WaitAsync(TestTimeout); + await WaitUntilAsync(() => hubContext.Sent.Count >= 2); + await cts.CancelAsync(); + await publisher.StopAsync(CancellationToken.None); + + IReadOnlyList<(string Method, object? Argument)> frames = hubContext.Frames; + Assert.Equal( + [AlarmsHub.ResetMessage, AlarmsHub.AlarmMessage, AlarmsHub.ResetMessage, AlarmsHub.AlarmMessage], + frames.Select(frame => frame.Method).Take(4)); + Assert.Equal(1, Assert.IsType(frames[0].Argument).Generation); + Assert.Equal(2, Assert.IsType(frames[2].Argument).Generation); + Assert.Same(first, frames[1].Argument); + Assert.Same(replayed, frames[3].Argument); + } + + /// + /// A broadcast that never completes — one client whose transport is wedged — is + /// abandoned at the send timeout, so the loop keeps draining the monitor + /// subscription instead of stalling until the monitor drops it. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ExecuteAsync_WhenABroadcastNeverCompletes_AbandonsItAndKeepsDraining() + { + AlarmFeedMessage stuck = BuildTransition(); + AlarmFeedMessage next = BuildActiveAlarm(); + ScriptedAlarmService alarmService = new([stuck, next]); + CapturingHubContext hubContext = new(); + hubContext.Proxy.StallFirstAlarm = true; + AlarmsHubPublisher publisher = CreatePublisher(alarmService, hubContext, TimeSpan.FromMilliseconds(50)); + + using CancellationTokenSource cts = new(); + await publisher.StartAsync(cts.Token).WaitAsync(TestTimeout); + await WaitUntilAsync(() => hubContext.Sent.Count >= 2); + await cts.CancelAsync(); + await publisher.StopAsync(CancellationToken.None); + + Assert.Equal("Tank02.Level.Lo", hubContext.Sent[1].ActiveAlarm.AlarmFullReference); + } + + private static AlarmsHubPublisher CreatePublisher( + IGatewayAlarmService alarmService, + CapturingHubContext hubContext, + TimeSpan sendTimeout, + bool showTagValues = true) + { + GatewayOptions gatewayOptions = new() + { + Dashboard = new DashboardOptions { ShowTagValues = showTagValues }, + }; + + return new AlarmsHubPublisher( + alarmService, + hubContext, + Options.Create(gatewayOptions), + NullLogger.Instance, + sendTimeout); + } + /// /// Starts the publisher over a scripted feed, waits until every scripted /// message has been broadcast, and stops it. @@ -149,16 +226,11 @@ public sealed class AlarmsHubPublisherTests { ScriptedAlarmService alarmService = new(messages); CapturingHubContext hubContext = new(); - GatewayOptions gatewayOptions = new() - { - Dashboard = new DashboardOptions { ShowTagValues = showTagValues }, - }; - - AlarmsHubPublisher publisher = new( + AlarmsHubPublisher publisher = CreatePublisher( alarmService, hubContext, - Options.Create(gatewayOptions), - NullLogger.Instance); + AlarmsHubPublisher.DefaultSendTimeout, + showTagValues); using CancellationTokenSource cts = new(); await publisher.StartAsync(cts.Token).WaitAsync(TestTimeout); @@ -221,12 +293,15 @@ public sealed class AlarmsHubPublisherTests } /// - /// Yields a scripted message list once and then stays open until cancelled, - /// so the publisher's reconnect loop never re-subscribes mid-test. + /// Yields one scripted batch per subscription. Every batch but the last ends its + /// stream — the monitor completing a subscription — so the publisher resubscribes; + /// the last batch stays open until cancelled, so the reconnect loop then settles. /// - /// The messages to yield, in order. - private sealed class ScriptedAlarmService(IReadOnlyList messages) : IGatewayAlarmService + /// The messages each successive subscription yields, in order. + private sealed class ScriptedAlarmService(params IReadOnlyList[] batches) : IGatewayAlarmService { + private int _subscriptions; + /// public GatewayAlarmMonitorState State => GatewayAlarmMonitorState.Monitoring; @@ -247,11 +322,24 @@ public sealed class AlarmsHubPublisherTests string? alarmFilterPrefix, [EnumeratorCancellation] CancellationToken cancellationToken) { + int index = Interlocked.Increment(ref _subscriptions) - 1; + IReadOnlyList messages = batches[Math.Min(index, batches.Length - 1)]; + if (index >= batches.Length) + { + // Past the script: stay open and quiet rather than replaying the last batch. + messages = []; + } + foreach (AlarmFeedMessage message in messages) { yield return message; } + if (index < batches.Length - 1) + { + yield break; + } + try { await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false); @@ -277,8 +365,14 @@ public sealed class AlarmsHubPublisherTests /// Gets the group manager. public IGroupManager Groups { get; } = new NoopGroupManager(); - /// Gets every message the publisher broadcast, in order. + /// Gets the capturing proxy every client lookup resolves to. + public CapturingClientProxy Proxy => _clients.GroupProxy; + + /// Gets every alarm message the publisher broadcast, in order. public IReadOnlyList Sent => _clients.GroupProxy.Sent; + + /// Gets every frame the publisher broadcast — alarm messages and reset markers — in order. + public IReadOnlyList<(string Method, object? Argument)> Frames => _clients.GroupProxy.Frames; } private sealed class CapturingHubClients : IHubClients @@ -334,6 +428,26 @@ public sealed class AlarmsHubPublisherTests private sealed class CapturingClientProxy : IClientProxy { private readonly List _sent = []; + private readonly List<(string Method, object? Argument)> _frames = []; + private bool _stalled; + + /// + /// Gets or sets a value indicating whether the first alarm send never completes + /// and ignores its cancellation token — a wedged client transport. + /// + public bool StallFirstAlarm { get; set; } + + /// Gets every frame sent through this proxy, in order. + public IReadOnlyList<(string Method, object? Argument)> Frames + { + get + { + lock (_sent) + { + return [.. _frames]; + } + } + } /// Gets every alarm message sent through this proxy, in order. public IReadOnlyList Sent @@ -347,22 +461,29 @@ public sealed class AlarmsHubPublisherTests } } - /// Records the broadcast message and completes synchronously. + /// Records the broadcast frame and completes synchronously unless stalled. /// The SignalR method name. /// The method arguments. /// Token to observe for cancellation. - /// A completed task. + /// A completed task, or a never-completing one for the stalled send. public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default) { - if (args.Length > 0 && args[0] is AlarmFeedMessage message) + bool stall = false; + lock (_sent) { - lock (_sent) + _frames.Add((method, args.Length > 0 ? args[0] : null)); + if (args.Length > 0 && args[0] is AlarmFeedMessage message) { _sent.Add(message); + if (StallFirstAlarm && !_stalled) + { + _stalled = true; + stall = true; + } } } - return Task.CompletedTask; + return stall ? new TaskCompletionSource().Task : Task.CompletedTask; } }