diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/BoundedHubSend.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/BoundedHubSend.cs new file mode 100644 index 0000000..4d14f37 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/BoundedHubSend.cs @@ -0,0 +1,52 @@ +using Microsoft.AspNetCore.SignalR; + +namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +/// +/// Time-bounded SignalR send shared by the dashboard hub publishers. A group or +/// broadcast send completes only once every target connection has accepted the +/// frame, so one remote viewer with a full transport buffer would otherwise hold +/// the publisher's single drain loop for as long as that viewer stays slow. +/// +internal static class BoundedHubSend +{ + /// + /// Sends as and waits at most + /// for it. The timeout cancels the send's own token, so + /// the lifetime manager abandons the pending write to the slow connection, and the + /// wait itself is bounded too, so a proxy that ignores the token cannot stall the + /// caller either. + /// + /// Target client proxy (a group or all connections). + /// Client method name. + /// The single message argument. + /// Upper bound on the send. + /// Host shutdown token; its cancellation propagates. + /// + /// when the send completed in time; + /// when it was abandoned at the timeout. + /// + /// was cancelled. + public static async Task SendAsync( + IClientProxy proxy, + string method, + object argument, + TimeSpan timeout, + CancellationToken stoppingToken) + { + using CancellationTokenSource bounded = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); + bounded.CancelAfter(timeout); + + try + { + await proxy.SendAsync(method, argument, bounded.Token) + .WaitAsync(bounded.Token) + .ConfigureAwait(false); + return true; + } + catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested) + { + return false; + } + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs index 909e7cc..5789d31 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs @@ -10,10 +10,11 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// Broadcasts MxEvents to the two dashboard audiences for a session: remote /// clients subscribed to the session's group, and /// in-process subscribers opened through -/// . Fire-and-forget: we -/// hand the send to the hub context and return immediately so the source gRPC -/// stream is never blocked. Errors are logged once and dropped — keeping the -/// SignalR mirror best-effort preserves the gRPC contract that exists today. +/// . +/// never blocks: both audiences are fed through bounded drop-oldest queues, so the +/// source gRPC stream is never held up by a slow dashboard viewer. Errors are +/// logged and dropped — keeping the SignalR mirror best-effort preserves the gRPC +/// contract. /// /// /// When MxGateway:Dashboard:ShowTagValues is false (the default), tag @@ -26,31 +27,45 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// ), while the redaction decides what a permitted /// subscriber sees — so the value-leak seam stays closed whatever the ACL /// admits. +/// +/// Hub delivery goes through one per watched session: a +/// -deep drop-oldest channel drained by a single task +/// that sends one event at a time, each send time-bounded. An earlier version fired +/// a group send per event and returned, so a slow remote viewer accumulated an +/// unbounded number of pending sends — each holding a cloned event — inside the +/// SignalR lifetime manager. The pump is started by the first event published while +/// a hub connection watches the session and torn down when +/// reports the last one gone. +/// /// -/// Hub context used to send to the session's group. -/// -/// Live-subscriber registry consulted before any per-event work is done. Both -/// audiences register here — hub connections by their SignalR connection id, -/// in-process subscriptions by a synthetic one — so the gate stays a single -/// source of truth. -/// -/// Gateway options supplying Dashboard:ShowTagValues. -/// Logger for best-effort mirror failures. -public sealed class DashboardEventBroadcaster( - IHubContext hubContext, - EventsHubViewerRegistry viewerRegistry, - IOptions options, - ILogger logger) : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber +public sealed class DashboardEventBroadcaster : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber, IDisposable { /// /// Queue depth per in-process subscriber. The consumer is a Blazor page /// rendering the newest handful of events, so a burst it cannot keep up with /// is dropped oldest-first rather than allowed to grow — same best-effort - /// contract the SignalR mirror already has. + /// contract the SignalR mirror has. /// private const int InProcessQueueCapacity = 256; - private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues; + /// + /// Queue depth of a session's hub-delivery pump. Same depth and drop policy as + /// the in-process feed: the remote viewer is a live view, so a burst it cannot + /// keep up with loses its oldest events rather than buffering without bound. + /// + internal const int HubQueueCapacity = 256; + + private static readonly TimeSpan DefaultHubSendTimeout = TimeSpan.FromSeconds(5); + + private readonly IHubContext _hubContext; + + private readonly EventsHubViewerRegistry _viewerRegistry; + + private readonly ILogger _logger; + + private readonly TimeSpan _hubSendTimeout; + + private readonly bool _showTagValues; private readonly object _syncRoot = new(); @@ -63,6 +78,9 @@ public sealed class DashboardEventBroadcaster( private readonly Dictionary _inProcessSubscribers = new(StringComparer.Ordinal); + /// Live hub-delivery pumps per session. Guarded by . + private readonly Dictionary _hubPumps = new(StringComparer.Ordinal); + /// /// Total live in-process subscribers, read without the lock so the common /// case — nobody has a session-details page open — never contends on it. @@ -70,6 +88,61 @@ public sealed class DashboardEventBroadcaster( /// private int _inProcessSubscriberCount; + private bool _disposed; + + /// Initializes a new instance of the class. + /// Hub context used to send to the session's group. + /// + /// Live-subscriber registry consulted before any per-event work is done. Both + /// audiences register here — hub connections by their SignalR connection id, + /// in-process subscriptions by a synthetic one — so the gate stays a single + /// source of truth. + /// + /// Gateway options supplying Dashboard:ShowTagValues. + /// Logger for best-effort mirror failures. + public DashboardEventBroadcaster( + IHubContext hubContext, + EventsHubViewerRegistry viewerRegistry, + IOptions options, + ILogger logger) + : this(hubContext, viewerRegistry, options, logger, DefaultHubSendTimeout) + { + } + + /// Initializes a new instance with a custom hub send timeout (test hook). + /// Hub context used to send to the session's group. + /// Live-subscriber registry. + /// Gateway options supplying Dashboard:ShowTagValues. + /// Logger for best-effort mirror failures. + /// Upper bound on one group send before it is abandoned. + internal DashboardEventBroadcaster( + IHubContext hubContext, + EventsHubViewerRegistry viewerRegistry, + IOptions options, + ILogger logger, + TimeSpan hubSendTimeout) + { + ArgumentNullException.ThrowIfNull(options); + _hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext)); + _viewerRegistry = viewerRegistry ?? throw new ArgumentNullException(nameof(viewerRegistry)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _hubSendTimeout = hubSendTimeout; + _showTagValues = options.Value.Dashboard.ShowTagValues; + _viewerRegistry.HubViewersReleased += OnHubViewersReleased; + } + + /// Gets the number of live hub-delivery pumps (one per hub-watched session). + internal int ActiveHubPumpCount + { + get + { + lock (_syncRoot) + { + return _hubPumps.Count; + } + } + } + /// public void Publish(string sessionId, MxEvent mxEvent) { @@ -81,46 +154,126 @@ public sealed class DashboardEventBroadcaster( // Every session's dashboard-mirror subscriber calls Publish for every event, // whether or not a browser is on that session's page. Without this gate the // steady state — no dashboard viewer at all — still paid a deep protobuf - // clone (redaction is on by default) plus a send to an empty SignalR group - // per event. Bail before both. - if (!viewerRegistry.HasViewers(sessionId)) + // clone (redaction is on by default) per event. Bail before it. + if (!_viewerRegistry.HasViewers(sessionId)) { return; } MxEvent outbound = _showTagValues ? mxEvent : RedactValues(mxEvent); - // In-process delivery first: it is synchronous, cannot throw, and must not be - // skipped by the early return the hub send's guard clause takes. DeliverInProcess(sessionId, outbound); - // Wrap the Task acquisition in a try/catch so a hypothetical synchronous throw - // from SendAsync (e.g. an implementation that throws before returning the Task) - // cannot escape Publish. The interface contract is never-throw; fire-and-forget. - Task send; + // In-process-only viewers need no hub send at all; the pump exists only + // while a remote connection is in the session's group. + if (_viewerRegistry.HasHubViewers(sessionId)) + { + EnqueueForHub(sessionId, outbound); + } + } + + /// + /// Stops every hub pump and detaches from the viewer registry. The singleton is + /// disposed with the host; a pump's in-flight send is bounded by its timeout. + /// + public void Dispose() + { + HubPump[] pumps; + lock (_syncRoot) + { + if (_disposed) + { + return; + } + + _disposed = true; + pumps = [.. _hubPumps.Values]; + _hubPumps.Clear(); + } + + _viewerRegistry.HubViewersReleased -= OnHubViewersReleased; + foreach (HubPump pump in pumps) + { + pump.Complete(); + } + } + + /// + /// Queues an event on the session's hub pump, starting the pump on first use. + /// The write happens under the lock so it cannot land on a pump that a + /// concurrent teardown has already completed. + /// + /// Session the event belongs to. + /// The event as the dashboard should see it. + private void EnqueueForHub(string sessionId, MxEvent outbound) + { + lock (_syncRoot) + { + if (_disposed) + { + return; + } + + if (!_hubPumps.TryGetValue(sessionId, out HubPump? pump)) + { + pump = new HubPump(this, sessionId, HubQueueCapacity); + _hubPumps[sessionId] = pump; + } + + pump.TryWrite(outbound); + } + } + + /// + /// Tears down a session's hub pump once its last hub viewer has gone. Re-checks + /// the registry under the lock: a viewer that re-subscribed in between keeps the + /// pump, and one that arrives after the teardown gets a fresh pump on the next + /// event. + /// + /// Session whose last hub viewer left. + private void OnHubViewersReleased(string sessionId) + { + HubPump? pump; + lock (_syncRoot) + { + if (_viewerRegistry.HasHubViewers(sessionId) || !_hubPumps.Remove(sessionId, out pump)) + { + return; + } + } + + pump.Complete(); + } + + /// + /// Sends one event to the session's group, bounded by the hub send timeout. + /// Never throws: a failed or abandoned send is logged and the pump moves on. + /// + /// Session whose group receives the event. + /// The event to send. + /// A task that completes when the send finished or was abandoned. + private async Task SendToHubAsync(string sessionId, MxEvent outbound) + { try { - send = hubContext.Clients - .Group(EventsHub.GroupName(sessionId)) - .SendAsync(EventsHub.EventMessage, outbound); + bool sent = await BoundedHubSend.SendAsync( + _hubContext.Clients.Group(EventsHub.GroupName(sessionId)), + EventsHub.EventMessage, + outbound, + _hubSendTimeout, + CancellationToken.None) + .ConfigureAwait(false); + if (!sent) + { + _logger.LogDebug( + "Dashboard event mirror to session {SessionId} timed out after {Timeout}; continuing.", + sessionId, + _hubSendTimeout); + } } catch (Exception ex) { - logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} threw synchronously.", sessionId); - return; - } - - if (!send.IsCompletedSuccessfully) - { - _ = send.ContinueWith( - t => - { - if (t.Exception is { } ex) - { - logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId); - } - }, - TaskScheduler.Default); + _logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId); } } @@ -133,7 +286,7 @@ public sealed class DashboardEventBroadcaster( // usable for a subscriber that has no SignalR connection behind it. The // "inproc-" prefix cannot collide with a SignalR connection id and makes the // origin obvious in a debugger. - string connectionId = "inproc-" + Guid.NewGuid().ToString("N"); + string connectionId = EventsHubViewerRegistry.InProcessConnectionIdPrefix + Guid.NewGuid().ToString("N"); InProcessSubscription subscription = new(this, sessionId, connectionId, InProcessQueueCapacity); // Register before the subscriber becomes a delivery target, exactly as @@ -142,7 +295,7 @@ public sealed class DashboardEventBroadcaster( // Publish's gate still reports the session unwatched, silently dropping events // it should receive. The cost of this order is at worst a redaction clone that // reaches nobody for the width of the window. - viewerRegistry.AddViewer(connectionId, sessionId); + _viewerRegistry.AddViewer(connectionId, sessionId); lock (_syncRoot) { @@ -227,12 +380,12 @@ public sealed class DashboardEventBroadcaster( } } - viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId); + _viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId); // The synthetic connection id is used once and never reconnects, so nothing // else will ever call ReleaseConnection for it; without this the registry // would retain an empty per-connection entry per subscription ever opened. - viewerRegistry.ReleaseConnection(subscription.ConnectionId); + _viewerRegistry.ReleaseConnection(subscription.ConnectionId); } /// @@ -328,4 +481,56 @@ public sealed class DashboardEventBroadcaster( _channel.Writer.TryComplete(); } } + + /// + /// One session's hub-delivery pump: a bounded drop-oldest channel and the single + /// task that drains it into the session's SignalR group, one bounded send at a + /// time. The drain is the only reader, so at most one send per session is ever + /// pending in the lifetime manager. + /// + private sealed class HubPump + { + private readonly Channel _channel; + + /// Initializes a new instance of the class and starts its drain. + /// Broadcaster whose send path the drain uses. + /// Session whose group the pump feeds. + /// Queue depth before the oldest queued event is dropped. + internal HubPump(DashboardEventBroadcaster owner, string sessionId, int capacity) + { + _channel = Channel.CreateBounded(new BoundedChannelOptions(capacity) + { + // DropOldest, not Wait: Publish runs on the gRPC event pipeline and + // must never block, and the newest events are the ones a live view wants. + FullMode = BoundedChannelFullMode.DropOldest, + SingleReader = true, + SingleWriter = false, + }); + + // Task.Run so the drain never starts inline on the publishing thread. + Completion = Task.Run(() => DrainAsync(owner, sessionId)); + } + + /// Gets the drain task; it completes once the channel is completed and emptied. + internal Task Completion { get; } + + /// Queues an event, dropping the oldest queued one when full. Never blocks. + /// The event to queue. + internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent); + + /// Completes the channel so the drain exits once it is empty. + internal void Complete() => _channel.Writer.TryComplete(); + + private async Task DrainAsync(DashboardEventBroadcaster owner, string sessionId) + { + ChannelReader reader = _channel.Reader; + while (await reader.WaitToReadAsync().ConfigureAwait(false)) + { + while (reader.TryRead(out MxEvent? next)) + { + await owner.SendToHubAsync(sessionId, next).ConfigureAwait(false); + } + } + } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs index ac66b00..965389a 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs @@ -19,11 +19,27 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// simply goes away never calls UnsubscribeSession; the hub's /// OnDisconnectedAsync releases everything the connection held. /// +/// +/// Hub viewers are also counted on their own, separately from the in-process +/// subscribers that register under an +/// id: runs one bounded hub-delivery pump +/// per session only while a remote connection is watching it, and tears that pump +/// down on . +/// /// public sealed class EventsHubViewerRegistry { + /// + /// Prefix of the synthetic connection ids in-process subscribers register under. + /// A SignalR connection id never carries it, so it is what tells the two + /// audiences apart for the hub-only count. + /// + public const string InProcessConnectionIdPrefix = "inproc-"; + private readonly ConcurrentDictionary _viewersBySession = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _hubViewersBySession = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _sessionsByConnection = new(StringComparer.Ordinal); @@ -53,8 +69,19 @@ public sealed class EventsHubViewerRegistry } _viewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1); + if (IsHubConnection(connectionId)) + { + _hubViewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1); + } } + /// + /// Raised with a session id when its last remote hub viewer leaves (unsubscribe + /// or disconnect). In-process subscribers do not raise it. Handlers run on the + /// releasing thread and must not throw or block. + /// + public event Action? HubViewersReleased; + /// /// Records that stopped watching /// . A removal with no matching @@ -75,7 +102,7 @@ public sealed class EventsHubViewerRegistry return; } - ReleaseSession(sessionId); + Release(connectionId, sessionId); } /// @@ -107,7 +134,7 @@ public sealed class EventsHubViewerRegistry // same detached set must not let the session be decremented twice. if (sessions.TryRemove(sessionId, out _)) { - ReleaseSession(sessionId); + Release(connectionId, sessionId); } } } @@ -120,32 +147,63 @@ public sealed class EventsHubViewerRegistry && _viewersBySession.TryGetValue(sessionId, out int count) && count > 0; + /// + /// Gets a value indicating whether any remote connection — + /// as opposed to an in-process subscriber — is watching the session. + /// + /// Session id to test. + /// when at least one hub connection is subscribed. + public bool HasHubViewers(string sessionId) => + !string.IsNullOrEmpty(sessionId) + && _hubViewersBySession.TryGetValue(sessionId, out int count) + && count > 0; + + private static bool IsHubConnection(string connectionId) => + !connectionId.StartsWith(InProcessConnectionIdPrefix, StringComparison.Ordinal); + + /// + /// Releases one subscription from both counts it was recorded in, and raises + /// when it was the session's last hub viewer. + /// + /// Connection that held the subscription. + /// Session id whose counts are released. + private void Release(string connectionId, string sessionId) + { + ReleaseSession(_viewersBySession, sessionId); + if (IsHubConnection(connectionId) && ReleaseSession(_hubViewersBySession, sessionId)) + { + HubViewersReleased?.Invoke(sessionId); + } + } + /// /// Decrements the session's viewer count, dropping the entry entirely at /// zero so the dictionary does not grow one key per session ever viewed. /// The compare-and-swap loop keeps the decrement correct against a /// concurrent on the same session. /// + /// The count map to decrement. /// Session id whose count is released. - private void ReleaseSession(string sessionId) + /// when this release dropped the count to zero. + private static bool ReleaseSession(ConcurrentDictionary counts, string sessionId) { while (true) { - if (!_viewersBySession.TryGetValue(sessionId, out int count)) + if (!counts.TryGetValue(sessionId, out int count)) { - return; + return false; } if (count <= 1) { - if (_viewersBySession.TryRemove(new KeyValuePair(sessionId, count))) + if (counts.TryRemove(new KeyValuePair(sessionId, count))) { - return; + return true; } } - else if (_viewersBySession.TryUpdate(sessionId, count - 1, count)) + else if (counts.TryUpdate(sessionId, count - 1, count)) { - return; + return false; } } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs index 9e5a0b5..d4b3688 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs @@ -12,7 +12,8 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard; /// MxGateway:Dashboard:ShowTagValues: tag values are stripped /// from the mirrored copy when the flag is off, present when it is on, and the /// shared source event is never mutated. Also verifies the viewer gate — the -/// mirror does no work at all for a session nobody is watching. +/// mirror does no work at all for a session nobody is watching, and that hub +/// delivery goes through one bounded drop-oldest pump per hub-watched session. /// public sealed class DashboardEventBroadcasterTests { @@ -22,11 +23,12 @@ public sealed class DashboardEventBroadcasterTests { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); + Assert.Equal(0, broadcaster.ActiveHubPumpCount); Assert.Equal(0, hubContext.SendCount); Assert.Null(hubContext.LastArgument); } @@ -38,40 +40,105 @@ public sealed class DashboardEventBroadcasterTests CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); viewers.AddViewer("conn-1", "session-2"); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); broadcaster.Publish("session-1", BuildEventWithValue()); + Assert.Equal(0, broadcaster.ActiveHubPumpCount); Assert.Equal(0, hubContext.SendCount); } - /// Once the last viewer leaves, the mirror stops sending again. + /// Once the last hub viewer leaves, the session's pump is torn down and sending stops. + /// A task that represents the asynchronous operation. [Fact] - public void Publish_AfterLastViewerLeaves_StopsSending() + public async Task Publish_AfterLastViewerLeaves_StopsSending() { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); viewers.AddViewer("conn-1", "session-1"); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); + await hubContext.WaitForSendCountAsync(1); + Assert.Equal(1, broadcaster.ActiveHubPumpCount); viewers.RemoveViewer("conn-1", "session-1"); + Assert.Equal(0, broadcaster.ActiveHubPumpCount); + broadcaster.Publish("session-1", BuildEventWithValue()); + Assert.Equal(0, broadcaster.ActiveHubPumpCount); Assert.Equal(1, hubContext.SendCount); } + /// + /// A hub viewer that cannot keep up costs a bounded queue, not an unbounded pile of + /// pending sends: while one send is stuck, further events drop oldest-first at the + /// pump's capacity, only one send is ever in flight, and the newest event survives. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Publish_WhenHubSendStalls_BuffersAtMostTheCapacityAndKeepsTheNewest() + { + CapturingHubContext hubContext = new(); + hubContext.Proxy.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + EventsHubViewerRegistry viewers = WatchedSession1(); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, viewers); + const int published = 2000; + + broadcaster.Publish("session-1", BuildSequencedEvent(0)); + await hubContext.WaitForSendCountAsync(1); + for (int i = 1; i < published; i++) + { + broadcaster.Publish("session-1", BuildSequencedEvent(i)); + } + + // One send stuck in flight, nothing else started behind it. + Assert.Equal(1, hubContext.SendCount); + Assert.Equal(1, hubContext.Proxy.MaxConcurrentSends); + + hubContext.Proxy.Gate.SetResult(); + await hubContext.WaitForLastItemHandleAsync(published - 1); + + // The stuck event plus at most one full queue — the rest were dropped oldest-first. + Assert.InRange(hubContext.SendCount, 2, DashboardEventBroadcaster.HubQueueCapacity + 1); + Assert.Equal(1, hubContext.Proxy.MaxConcurrentSends); + } + + /// + /// A send that never completes is abandoned at the send timeout, so one stuck + /// connection cannot stop the pump delivering later events. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Publish_WhenHubSendNeverCompletes_AbandonsItAtTheTimeoutAndContinues() + { + CapturingHubContext hubContext = new(); + hubContext.Proxy.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + hubContext.Proxy.IgnoreCancellation = true; + using DashboardEventBroadcaster broadcaster = Create( + hubContext, + showTagValues: true, + WatchedSession1(), + hubSendTimeout: TimeSpan.FromMilliseconds(50)); + + broadcaster.Publish("session-1", BuildSequencedEvent(1)); + broadcaster.Publish("session-1", BuildSequencedEvent(2)); + + await hubContext.WaitForSendCountAsync(2); + } + /// Values are stripped from the mirror when ShowTagValues is off; metadata survives. [Fact] - public void Publish_WhenShowTagValuesFalse_RedactsValuesButKeepsMetadata() + /// A task that represents the asynchronous operation. + public async Task Publish_WhenShowTagValuesFalse_RedactsValuesButKeepsMetadata() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); + await hubContext.WaitForSendCountAsync(1); MxEvent sent = Assert.IsType(hubContext.LastArgument); Assert.Null(sent.Value); @@ -88,13 +155,15 @@ public sealed class DashboardEventBroadcasterTests /// Redaction applies to a clone, so the shared source event keeps its values. [Fact] - public void Publish_WhenShowTagValuesFalse_DoesNotMutateSourceEvent() + /// A task that represents the asynchronous operation. + public async Task Publish_WhenShowTagValuesFalse_DoesNotMutateSourceEvent() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); + await hubContext.WaitForSendCountAsync(1); // The redaction must apply to a clone; the shared source keeps its values. Assert.NotNull(source.Value); @@ -106,13 +175,15 @@ public sealed class DashboardEventBroadcasterTests /// Values pass through unredacted when ShowTagValues is on. [Fact] - public void Publish_WhenShowTagValuesTrue_KeepsValues() + /// A task that represents the asynchronous operation. + public async Task Publish_WhenShowTagValuesTrue_KeepsValues() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, WatchedSession1()); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); + await hubContext.WaitForSendCountAsync(1); MxEvent sent = Assert.IsType(hubContext.LastArgument); Assert.NotNull(sent.Value); @@ -125,16 +196,18 @@ public sealed class DashboardEventBroadcasterTests /// An in-process subscriber gets the same redacted clone the hub group gets, /// and the shared source event is still left untouched. /// + /// A task that represents the asynchronous operation. [Fact] - public void Subscribe_WhenShowTagValuesFalse_DeliversRedactedCloneWithoutMutatingSource() + public async Task Subscribe_WhenShowTagValuesFalse_DeliversRedactedCloneWithoutMutatingSource() { CapturingHubContext hubContext = new(); - EventsHubViewerRegistry viewers = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + EventsHubViewerRegistry viewers = WatchedSession1(); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); + await hubContext.WaitForSendCountAsync(1); MxEvent received = ReadOne(subscription); Assert.Null(received.Value); @@ -153,17 +226,16 @@ public sealed class DashboardEventBroadcasterTests Assert.NotNull(source.OnAlarmTransition.LimitValue); } - /// An in-process subscription opens the viewer gate the same way a hub client does. + /// + /// An in-process subscription opens the viewer gate the same way a hub client does, + /// but starts no hub pump: with no remote connection there is nobody to send to. + /// [Fact] public void Subscribe_OpensTheViewerGate() { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); - - broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(0, hubContext.SendCount); - Assert.Null(hubContext.LastArgument); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); @@ -171,8 +243,9 @@ public sealed class DashboardEventBroadcasterTests broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); - Assert.NotNull(hubContext.LastArgument); + Assert.Equal("session-1", ReadOne(subscription).SessionId); + Assert.Equal(0, broadcaster.ActiveHubPumpCount); + Assert.Equal(0, hubContext.SendCount); } /// Disposing the last in-process subscription restores the no-viewers short-circuit. @@ -181,11 +254,10 @@ public sealed class DashboardEventBroadcasterTests { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); Assert.Equal("session-1", ReadOne(subscription).SessionId); subscription.Dispose(); @@ -194,26 +266,28 @@ public sealed class DashboardEventBroadcasterTests broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); + Assert.Equal(0, hubContext.SendCount); Assert.False(subscription.Reader.TryRead(out _)); } /// Hub viewers and in-process subscribers are audiences of their own session only. + /// A task that represents the asynchronous operation. [Fact] - public void Publish_DeliversOnlyToTheSubscribedSessionsAudience() + public async Task Publish_DeliversOnlyToTheSubscribedSessionsAudience() { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); viewers.AddViewer("conn-1", "session-1"); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-2"); // The hub viewer's session must not spill into the in-process feed. broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); + await hubContext.WaitForSendCountAsync(1); Assert.False(subscription.Reader.TryRead(out _)); + // The in-process-only session reaches its subscriber and starts no hub pump. broadcaster.Publish("session-2", BuildEventWithValue("session-2")); Assert.Equal("session-2", ReadOne(subscription).SessionId); @@ -221,7 +295,8 @@ public sealed class DashboardEventBroadcasterTests // A session with no audience at all still short-circuits. broadcaster.Publish("session-3", BuildEventWithValue("session-3")); - Assert.Equal(2, hubContext.SendCount); + Assert.Equal(1, broadcaster.ActiveHubPumpCount); + Assert.Equal(1, hubContext.SendCount); } /// Disposing twice is a no-op and cannot release a sibling subscription's registration. @@ -230,7 +305,7 @@ public sealed class DashboardEventBroadcasterTests { CapturingHubContext hubContext = new(); EventsHubViewerRegistry viewers = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); IDashboardEventSubscription first = broadcaster.Subscribe("session-1"); using IDashboardEventSubscription second = broadcaster.Subscribe("session-1"); @@ -241,7 +316,6 @@ public sealed class DashboardEventBroadcasterTests broadcaster.Publish("session-1", BuildEventWithValue()); - Assert.Equal(1, hubContext.SendCount); Assert.False(first.Reader.TryRead(out _)); Assert.Equal("session-1", ReadOne(second).SessionId); } @@ -258,7 +332,8 @@ public sealed class DashboardEventBroadcasterTests private static DashboardEventBroadcaster Create( CapturingHubContext hubContext, bool showTagValues, - EventsHubViewerRegistry viewers) + EventsHubViewerRegistry viewers, + TimeSpan? hubSendTimeout = null) { GatewayOptions gatewayOptions = new() { @@ -269,7 +344,21 @@ public sealed class DashboardEventBroadcasterTests hubContext, viewers, Options.Create(gatewayOptions), - NullLogger.Instance); + NullLogger.Instance, + hubSendTimeout ?? TimeSpan.FromSeconds(30)); + } + + /// Builds a value-less event whose item handle carries its publish order. + /// Publish order stamped into . + /// The event. + private static MxEvent BuildSequencedEvent(int sequence) + { + return new MxEvent + { + Family = MxEventFamily.OnDataChange, + SessionId = "session-1", + ItemHandle = sequence, + }; } /// A registry with one hub connection watching session-1. @@ -313,81 +402,157 @@ public sealed class DashboardEventBroadcasterTests /// Gets the group manager. public IGroupManager Groups { get; } = new NoopGroupManager(); + /// Gets the capturing proxy every client lookup resolves to. + public CapturingClientProxy Proxy => _clients.Proxy; + /// Gets the first argument of the most recent send call. - public object? LastArgument => _clients.GroupProxy.LastArgument; + public object? LastArgument => _clients.Proxy.LastArgument; /// Gets the number of send calls this fake has observed. - public int SendCount => _clients.GroupProxy.SendCount; + public int SendCount => _clients.Proxy.SendCount; + + /// Waits until at least sends have started. + /// The send count to wait for. + /// A task that completes once the count is reached. + public Task WaitForSendCountAsync(int count) => + WaitUntilAsync(() => SendCount >= count, $"{count} sends (saw {SendCount})"); + + /// Waits until a send carrying the given item handle has started. + /// Item handle of the event to wait for. + /// A task that completes once the event was sent. + public Task WaitForLastItemHandleAsync(int itemHandle) => + WaitUntilAsync( + () => LastArgument is MxEvent sent && sent.ItemHandle == itemHandle, + $"a send of item handle {itemHandle}"); + + private static async Task WaitUntilAsync(Func condition, string description) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(10); + while (!condition()) + { + if (DateTime.UtcNow > deadline) + { + Assert.Fail($"Timed out waiting for {description}."); + } + + await Task.Delay(5); + } + } } private sealed class CapturingHubClients : IHubClients { /// Gets the capturing client proxy shared by this fake. - public CapturingClientProxy GroupProxy { get; } = new(); + public CapturingClientProxy Proxy { get; } = new(); /// Gets the shared capturing proxy for all connections. - public IClientProxy All => GroupProxy; + public IClientProxy All => Proxy; /// Returns the shared capturing proxy, ignoring exclusions. /// Ignored. /// The shared capturing proxy. - public IClientProxy AllExcept(IReadOnlyList excludedConnectionIds) => GroupProxy; + public IClientProxy AllExcept(IReadOnlyList excludedConnectionIds) => Proxy; /// Returns the shared capturing proxy for any connection. /// Ignored. /// The shared capturing proxy. - public IClientProxy Client(string connectionId) => GroupProxy; + public IClientProxy Client(string connectionId) => Proxy; /// Returns the shared capturing proxy for any connection set. /// Ignored. /// The shared capturing proxy. - public IClientProxy Clients(IReadOnlyList connectionIds) => GroupProxy; + public IClientProxy Clients(IReadOnlyList connectionIds) => Proxy; /// Returns the shared capturing proxy for any group. /// Ignored. /// The shared capturing proxy. - public IClientProxy Group(string groupName) => GroupProxy; + public IClientProxy Group(string groupName) => Proxy; /// Returns the shared capturing proxy for any group, ignoring exclusions. /// Ignored. /// Ignored. /// The shared capturing proxy. - public IClientProxy GroupExcept(string groupName, IReadOnlyList excludedConnectionIds) => GroupProxy; + public IClientProxy GroupExcept(string groupName, IReadOnlyList excludedConnectionIds) => Proxy; /// Returns the shared capturing proxy for any group set. /// Ignored. /// The shared capturing proxy. - public IClientProxy Groups(IReadOnlyList groupNames) => GroupProxy; + public IClientProxy Groups(IReadOnlyList groupNames) => Proxy; /// Returns the shared capturing proxy for any user. /// Ignored. /// The shared capturing proxy. - public IClientProxy User(string userId) => GroupProxy; + public IClientProxy User(string userId) => Proxy; /// Returns the shared capturing proxy for any user set. /// Ignored. /// The shared capturing proxy. - public IClientProxy Users(IReadOnlyList userIds) => GroupProxy; + public IClientProxy Users(IReadOnlyList userIds) => Proxy; } private sealed class CapturingClientProxy : IClientProxy { + private int _sendCount; + private int _inFlight; + private int _maxConcurrentSends; + private object? _lastArgument; + + /// + /// Gets or sets a gate every send waits on before completing; null completes + /// sends synchronously. Set it before the first publish. + /// + public TaskCompletionSource? Gate { get; set; } + + /// Gets or sets a value indicating whether a gated send ignores its cancellation token. + public bool IgnoreCancellation { get; set; } + /// Gets the first argument of the most recent send call. - public object? LastArgument { get; private set; } + public object? LastArgument => Volatile.Read(ref _lastArgument); /// Gets the number of send calls made through this proxy. - public int SendCount { get; private set; } + public int SendCount => Volatile.Read(ref _sendCount); - /// Records the send call arguments and completes synchronously. + /// Gets the highest number of sends that were in flight at once. + public int MaxConcurrentSends => Volatile.Read(ref _maxConcurrentSends); + + /// Records the send call arguments and completes once the gate (if any) opens. /// The SignalR method name. /// The method arguments. /// Token to observe for cancellation. - /// A completed task. + /// A task that completes when the send is released. public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default) { - SendCount++; - LastArgument = args.Length > 0 ? args[0] : null; - return Task.CompletedTask; + Volatile.Write(ref _lastArgument, args.Length > 0 ? args[0] : null); + Interlocked.Increment(ref _sendCount); + + TaskCompletionSource? gate = Gate; + if (gate is null || gate.Task.IsCompleted) + { + return Task.CompletedTask; + } + + return TrackAsync(gate.Task, IgnoreCancellation ? CancellationToken.None : cancellationToken); + } + + private async Task TrackAsync(Task gate, CancellationToken cancellationToken) + { + int inFlight = Interlocked.Increment(ref _inFlight); + int observed; + do + { + observed = Volatile.Read(ref _maxConcurrentSends); + } + while (inFlight > observed + && Interlocked.CompareExchange(ref _maxConcurrentSends, inFlight, observed) != observed); + + try + { + await gate.WaitAsync(cancellationToken); + } + finally + { + Interlocked.Decrement(ref _inFlight); + } } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs index acd5712..17384d8 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs @@ -164,4 +164,48 @@ public sealed class EventsHubViewerRegistryTests Assert.False(registry.HasViewers("session-1")); } + + /// In-process subscribers count as viewers but not as hub viewers. + [Fact] + public void HasHubViewers_CountsOnlyRemoteConnections() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1"); + + Assert.True(registry.HasViewers("session-1")); + Assert.False(registry.HasHubViewers("session-1")); + + registry.AddViewer("conn-1", "session-1"); + + Assert.True(registry.HasHubViewers("session-1")); + } + + /// + /// The released event fires once, when the last hub viewer leaves — whether it + /// unsubscribes or disconnects — and never for in-process subscribers. + /// + [Fact] + public void HubViewersReleased_FiresWhenTheLastHubViewerLeaves() + { + EventsHubViewerRegistry registry = new(); + List released = []; + registry.HubViewersReleased += released.Add; + + registry.AddViewer("conn-1", "session-1"); + registry.AddViewer("conn-2", "session-1"); + registry.AddViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1"); + + registry.RemoveViewer("conn-1", "session-1"); + Assert.Empty(released); + + registry.ReleaseConnection("conn-2"); + Assert.Equal(["session-1"], released); + Assert.False(registry.HasHubViewers("session-1")); + + // The in-process subscriber still keeps the session watched, and leaving raises nothing. + Assert.True(registry.HasViewers("session-1")); + registry.RemoveViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1"); + Assert.Equal(["session-1"], released); + } }