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.
2026-10-22 06:14:53 -04:00
parent f1db002b87
commit 3688fa19e6
4 changed files with 270 additions and 43 deletions
@@ -0,0 +1,14 @@
+namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
+
+/// <summary>
+/// Payload of the <see cref="AlarmsHub.ResetMessage"/> 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 <c>active_alarm</c> rows up to the next
+/// <c>snapshot_complete</c>.
+/// </summary>
+/// <param name="Generation">
+/// Monotonic subscription counter for this gateway process, starting at 1. Lets a
+/// client tell a resubscribe from a duplicate delivery.
+/// </param>
+public sealed record AlarmFeedReset(long Generation);
@@ -23,11 +23,19 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
 [Authorize(Policy = DashboardAuthenticationDefaults.AlarmsPolicy)]
 public sealed class AlarmsHub : Hub
 {
+    /// <summary>Group every admitted connection joins on connect.</summary>
    public const string AllAlarmsGroup = "__alarms__";

    /// <summary>Method name used to push <c>AlarmFeedMessage</c> values to clients.</summary>
    public const string AlarmMessage = "AlarmFeed";

+    /// <summary>
+    /// Method name of the <see cref="AlarmFeedReset"/> 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.
+    /// </summary>
+    public const string ResetMessage = "AlarmFeedReset";
+
    /// <inheritdoc />
    public override async Task OnConnectedAsync()
    {
@@ -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.
 /// </para>
+/// <para>
+/// Each broadcast is time-bounded (<see cref="DefaultSendTimeout"/>). 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.
+/// </para>
+/// <para>
+/// Every subscription — the first and each resubscribe — is preceded by an
+/// <see cref="AlarmsHub.ResetMessage"/> frame. The monitor primes every new
+/// subscription with a full snapshot, so a client that keeps accumulating
+/// <c>active_alarm</c> 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.
+/// </para>
 /// </remarks>
-/// <param name="alarmService">The gateway's central alarm feed.</param>
-/// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
-/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
-/// <param name="logger">Logger for best-effort broadcast failures.</param>
-public sealed class AlarmsHubPublisher(
-    IGatewayAlarmService alarmService,
-    IHubContext<AlarmsHub> hubContext,
-    IOptions<GatewayOptions> options,
-    ILogger<AlarmsHubPublisher> logger) : BackgroundService
+public sealed class AlarmsHubPublisher : BackgroundService
 {
-    private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues;
+    /// <summary>Upper bound on one alarm broadcast before it is abandoned.</summary>
+    internal static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(2);
+
+    private readonly IGatewayAlarmService _alarmService;
+    private readonly IHubContext<AlarmsHub> _hubContext;
+    private readonly ILogger<AlarmsHubPublisher> _logger;
+    private readonly TimeSpan _sendTimeout;
+    private readonly bool _showTagValues;
+    private long _subscriptionGeneration;
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmsHubPublisher"/> class.</summary>
+    /// <param name="alarmService">The gateway's central alarm feed.</param>
+    /// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
+    /// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
+    /// <param name="logger">Logger for best-effort broadcast failures.</param>
+    public AlarmsHubPublisher(
+        IGatewayAlarmService alarmService,
+        IHubContext<AlarmsHub> hubContext,
+        IOptions<GatewayOptions> options,
+        ILogger<AlarmsHubPublisher> logger)
+        : this(alarmService, hubContext, options, logger, DefaultSendTimeout)
+    {
+    }
+
+    /// <summary>Initializes a new instance with a custom send timeout (test hook).</summary>
+    /// <param name="alarmService">The gateway's central alarm feed.</param>
+    /// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
+    /// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
+    /// <param name="logger">Logger for best-effort broadcast failures.</param>
+    /// <param name="sendTimeout">Upper bound on one broadcast.</param>
+    internal AlarmsHubPublisher(
+        IGatewayAlarmService alarmService,
+        IHubContext<AlarmsHub> hubContext,
+        IOptions<GatewayOptions> options,
+        ILogger<AlarmsHubPublisher> 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;
+    }

    /// <inheritdoc />
    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(
        }
    }

+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <param name="method">Client method name.</param>
+    /// <param name="argument">The frame payload.</param>
+    /// <param name="stoppingToken">Host shutdown token.</param>
+    /// <returns>A task that completes when the send finished or was abandoned.</returns>
+    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.");
+        }
+    }
+
    /// <summary>
    /// Produces a copy of <paramref name="source"/> with the alarm value fields
    /// cleared, leaving every other field — reference, severity, state, category,
@@ -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
 /// <see cref="AlarmFeedMessage"/> — shared with the gRPC <c>StreamAlarms</c>
-/// 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.
 /// </summary>
 public sealed class AlarmsHubPublisherTests
 {
@@ -136,6 +138,81 @@ public sealed class AlarmsHubPublisherTests
        Assert.True(hubContext.Sent[2].SnapshotStatus.Truncated);
    }

+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<AlarmFeedReset>(frames[0].Argument).Generation);
+        Assert.Equal(2, Assert.IsType<AlarmFeedReset>(frames[2].Argument).Generation);
+        Assert.Same(first, frames[1].Argument);
+        Assert.Same(replayed, frames[3].Argument);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<AlarmsHubPublisher>.Instance,
+            sendTimeout);
+    }
+
    /// <summary>
    /// 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<AlarmsHubPublisher>.Instance);
+            AlarmsHubPublisher.DefaultSendTimeout,
+            showTagValues);

        using CancellationTokenSource cts = new();
        await publisher.StartAsync(cts.Token).WaitAsync(TestTimeout);
@@ -221,12 +293,15 @@ public sealed class AlarmsHubPublisherTests
    }

    /// <summary>
-    /// 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.
    /// </summary>
-    /// <param name="messages">The messages to yield, in order.</param>
-    private sealed class ScriptedAlarmService(IReadOnlyList<AlarmFeedMessage> messages) : IGatewayAlarmService
+    /// <param name="batches">The messages each successive subscription yields, in order.</param>
+    private sealed class ScriptedAlarmService(params IReadOnlyList<AlarmFeedMessage>[] batches) : IGatewayAlarmService
    {
+        private int _subscriptions;
+
        /// <inheritdoc />
        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<AlarmFeedMessage> 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
        /// <summary>Gets the group manager.</summary>
        public IGroupManager Groups { get; } = new NoopGroupManager();

-        /// <summary>Gets every message the publisher broadcast, in order.</summary>
+        /// <summary>Gets the capturing proxy every client lookup resolves to.</summary>
+        public CapturingClientProxy Proxy => _clients.GroupProxy;
+
+        /// <summary>Gets every alarm message the publisher broadcast, in order.</summary>
        public IReadOnlyList<AlarmFeedMessage> Sent => _clients.GroupProxy.Sent;
+
+        /// <summary>Gets every frame the publisher broadcast — alarm messages and reset markers — in order.</summary>
+        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<AlarmFeedMessage> _sent = [];
+        private readonly List<(string Method, object? Argument)> _frames = [];
+        private bool _stalled;
+
+        /// <summary>
+        /// Gets or sets a value indicating whether the first alarm send never completes
+        /// and ignores its cancellation token — a wedged client transport.
+        /// </summary>
+        public bool StallFirstAlarm { get; set; }
+
+        /// <summary>Gets every frame sent through this proxy, in order.</summary>
+        public IReadOnlyList<(string Method, object? Argument)> Frames
+        {
+            get
+            {
+                lock (_sent)
+                {
+                    return [.. _frames];
+                }
+            }
+        }

        /// <summary>Gets every alarm message sent through this proxy, in order.</summary>
        public IReadOnlyList<AlarmFeedMessage> Sent
@@ -347,22 +461,29 @@ public sealed class AlarmsHubPublisherTests
            }
        }

-        /// <summary>Records the broadcast message and completes synchronously.</summary>
+        /// <summary>Records the broadcast frame and completes synchronously unless stalled.</summary>
        /// <param name="method">The SignalR method name.</param>
        /// <param name="args">The method arguments.</param>
        /// <param name="cancellationToken">Token to observe for cancellation.</param>
-        /// <returns>A completed task.</returns>
+        /// <returns>A completed task, or a never-completing one for the stalled send.</returns>
        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;
        }
    }