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; } }