From 6c5218913bbd5c466bd88d01ee6bcb630b6bee07 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:07:33 -0400 Subject: [PATCH] =?UTF-8?q?perf(dashboard):=20gate=20event=20mirror=20on?= =?UTF-8?q?=20live=20viewers=20=E2=80=94=20no=20clone,=20no=20send=20for?= =?UTF-8?q?=20unwatched=20sessions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DashboardEventBroadcaster.Publish ran a deep protobuf Clone (redaction is on by default) and a group SendAsync for every event of every session, before anything checked whether a dashboard client was actually watching. In the steady state the session:{id} group is empty, so that work was thrown away per event. SignalR does not expose group membership, so EventsHub now mirrors its own add/remove into a singleton EventsHubViewerRegistry, and OnDisconnectedAsync releases everything a dropped connection held (SessionDetailsPage disposes the connection rather than unsubscribing). Publish returns early when the session has no viewers, before the redaction clone. Watched sessions behave exactly as before. Lazy mirror-lease start/stop was deliberately not attempted — it entangles the dashboard with SessionEventDistributor subscribe lifetime for no saving beyond this gate; recorded in docs/GatewayDashboardDesign.md. --- docs/GatewayDashboardDesign.md | 27 ++- .../DashboardServiceCollectionExtensions.cs | 3 + .../Hubs/DashboardEventBroadcaster.cs | 17 ++ .../Dashboard/Hubs/EventsHub.cs | 38 +++- .../Dashboard/Hubs/EventsHubViewerRegistry.cs | 147 ++++++++++++++++ .../DashboardEventBroadcasterTests.cs | 78 ++++++++- .../Dashboard/EventsHubViewerRegistryTests.cs | 165 ++++++++++++++++++ 7 files changed, 465 insertions(+), 10 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs diff --git a/docs/GatewayDashboardDesign.md b/docs/GatewayDashboardDesign.md index e7fb554..5267fdc 100644 --- a/docs/GatewayDashboardDesign.md +++ b/docs/GatewayDashboardDesign.md @@ -167,7 +167,7 @@ bearer). Each hub class is `[Authorize(Policy = HubClientsPolicy)]`. |---|---|---|---|---| | `DashboardSnapshotHub` | `/hubs/snapshot` | `DashboardSnapshotPublisher` (BackgroundService consuming `IDashboardSnapshotService.WatchSnapshotsAsync`) | `DashboardSnapshot` | Sent to all connected clients on every snapshot tick; new connections receive the current snapshot synchronously in `OnConnectedAsync`. | | `AlarmsHub` | `/hubs/alarms` | `AlarmsHubPublisher` (BackgroundService consuming `IGatewayAlarmService.StreamAsync(filter: null)`) | `AlarmFeedMessage` (`active_alarm` / `snapshot_complete` / `transition`) | Connected clients auto-join `__alarms__`; all clients receive every message. Publisher auto-reconnects every 5s on stream faults. | -| `EventsHub` | `/hubs/events` | `DashboardEventBroadcaster` invoked by each session's internal dashboard-mirror subscriber on its `SessionEventDistributor` (registered when the session becomes Ready) | `MxEvent` | Clients call `SubscribeSession(sessionId)` to join `session:{id}`. The dashboard is a first-class distributor subscriber, so it receives the session's events whether or not a gRPC client is streaming. It sees RAW session events — not the per-gRPC-subscriber `AfterWorkerSequence` filtering that `EventStreamService` applies at its own boundary — because the dashboard is a separate LDAP-authenticated monitoring view meant to show the session's full event activity. Tag values are stripped from the mirrored `MxEvent` copy by `DashboardEventBroadcaster` when `Dashboard:ShowTagValues` is false (the default) — event metadata (tag reference, quality, status, timestamps) still renders, but the value fields are blanked, so no value leaks through this seam. The per-session hub ACL that would scope a Viewer to specific sessions is still outstanding (SEC-25 / remediation roadmap item 12); the value redaction is the near-term hardening that closes the value-leak seam independently of that ACL. | +| `EventsHub` | `/hubs/events` | `DashboardEventBroadcaster` invoked by each session's internal dashboard-mirror subscriber on its `SessionEventDistributor` (registered when the session becomes Ready) | `MxEvent` | Clients call `SubscribeSession(sessionId)` to join `session:{id}`, which also registers them in `EventsHubViewerRegistry` — the mirror is gated on that registry (see "Mirror gating" below). The dashboard is a first-class distributor subscriber, so it receives the session's events whether or not a gRPC client is streaming. It sees RAW session events — not the per-gRPC-subscriber `AfterWorkerSequence` filtering that `EventStreamService` applies at its own boundary — because the dashboard is a separate LDAP-authenticated monitoring view meant to show the session's full event activity. Tag values are stripped from the mirrored `MxEvent` copy by `DashboardEventBroadcaster` when `Dashboard:ShowTagValues` is false (the default) — event metadata (tag reference, quality, status, timestamps) still renders, but the value fields are blanked, so no value leaks through this seam. The per-session hub ACL that would scope a Viewer to specific sessions is still outstanding (SEC-25 / remediation roadmap item 12); the value redaction is the near-term hardening that closes the value-leak seam independently of that ACL. | `DashboardPageBase` opens a `DashboardSnapshotHub` connection via the connection factory in `OnInitializedAsync`, seeds `Snapshot` synchronously from @@ -191,6 +191,31 @@ Avoid pushing every MXAccess data-change event into a wider broadcast group. The current design routes events strictly through `session:{id}` groups; the snapshot hub continues to carry aggregate event counters and rates. +### Mirror gating + +Each session's dashboard-mirror subscriber calls +`DashboardEventBroadcaster.Publish` for every event the session produces, +independently of whether any browser is watching that session. SignalR does not +expose group membership, so the broadcaster cannot ask whether `session:{id}` is +empty. `EventsHubViewerRegistry` (singleton) supplies that answer: `EventsHub` +mirrors its own `AddToGroup` / `RemoveFromGroup` calls into it, and +`OnDisconnectedAsync` releases every subscription a dropped connection held — the +only reliable signal for a browser tab that closes without unsubscribing. +`Publish` returns immediately when `HasViewers(sessionId)` is false, **before** +the redaction clone. That matters because redaction is on by default +(`Dashboard:ShowTagValues` false), so the unwatched steady state — nobody on any +session-details page — previously paid a deep protobuf clone plus a send to an +empty group for every event of every session. Behaviour for a watched session is +unchanged. + +The mirror subscriber itself is still registered on the `SessionEventDistributor` +for the session's whole lifetime; only the per-event work is gated. Starting and +stopping the mirror lease lazily with the first and last viewer was considered +and deliberately not done — it entangles the dashboard with distributor +subscribe/unsubscribe lifetime (and with the replay/sequence bookkeeping that +attaching a subscriber mid-stream implies) for no additional saving beyond the +clone and send this gate already removes. + ## Pages ### Dashboard home diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs index 658d42e..c28f859 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs @@ -47,6 +47,9 @@ public static class DashboardServiceCollectionExtensions services.AddSingleton(); services.AddScoped(); services.AddScoped(); + // Singleton: EventsHub instances are transient (one per hub invocation), so the + // subscriber bookkeeping they share with the broadcaster must outlive them. + services.AddSingleton(); services.AddSingleton(); services.AddHostedService(); services.AddHostedService(); 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 b81125e..33597d6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs @@ -21,8 +21,15 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// the mirror independently of the still-outstanding per-session hub ACL /// (see ). /// +/// Hub context used to send to the session's group. +/// +/// Live-subscriber registry consulted before any per-event work is done. +/// +/// Gateway options supplying Dashboard:ShowTagValues. +/// Logger for best-effort mirror failures. public sealed class DashboardEventBroadcaster( IHubContext hubContext, + EventsHubViewerRegistry viewerRegistry, IOptions options, ILogger logger) : IDashboardEventBroadcaster { @@ -36,6 +43,16 @@ public sealed class DashboardEventBroadcaster( return; } + // 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)) + { + return; + } + MxEvent outbound = _showTagValues ? mxEvent : RedactValues(mxEvent); // Wrap the Task acquisition in a try/catch so a hypothetical synchronous throw diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHub.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHub.cs index 1c5863c..53c5c1c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHub.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHub.cs @@ -9,8 +9,14 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// session; sends messages to /// session:{id} as events arrive from the live gRPC stream. /// +/// +/// Group membership is mirrored into +/// because SignalR does not expose it, and the broadcaster consults the +/// registry to skip all mirror work for sessions nobody is watching. +/// +/// Registry tracking which sessions have live subscribers. [Authorize(Policy = DashboardAuthenticationDefaults.HubClientsPolicy)] -public sealed class EventsHub : Hub +public sealed class EventsHub(EventsHubViewerRegistry viewerRegistry) : Hub { /// Method name used to push individual MxEvent values to clients. public const string EventMessage = "MxEvent"; @@ -55,19 +61,43 @@ public sealed class EventsHub : Hub return Task.CompletedTask; } + // Register before joining the group: the reverse order would leave a window + // in which this connection is a group member but the broadcaster's gate still + // reports the session unwatched, silently dropping events it should receive. + viewerRegistry.AddViewer(Context.ConnectionId, sessionId); + return Groups.AddToGroupAsync(Context.ConnectionId, GroupName(sessionId)); } /// Unsubscribes the calling SignalR connection from the per-session events group. /// Session id to unsubscribe the caller from. /// A task representing the unsubscription operation. - public Task UnsubscribeSession(string sessionId) + public async Task UnsubscribeSession(string sessionId) { if (string.IsNullOrWhiteSpace(sessionId)) { - return Task.CompletedTask; + return; } - return Groups.RemoveFromGroupAsync(Context.ConnectionId, GroupName(sessionId)); + // Leave the group first, deregister after — the mirror stays enabled for the + // brief overlap rather than dropping events still owed to other subscribers. + await Groups.RemoveFromGroupAsync(Context.ConnectionId, GroupName(sessionId)).ConfigureAwait(false); + + viewerRegistry.RemoveViewer(Context.ConnectionId, sessionId); + } + + /// + /// Releases every session subscription the dropped connection held. A browser + /// tab that closes never calls , so without + /// this the session would look watched forever and the mirror would keep + /// cloning and sending events to an empty group. + /// + /// The exception that terminated the connection, if any. + /// A task representing the disconnect handling. + public override Task OnDisconnectedAsync(Exception? exception) + { + viewerRegistry.ReleaseConnection(Context.ConnectionId); + + return base.OnDisconnectedAsync(exception); } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs new file mode 100644 index 0000000..0607dc3 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs @@ -0,0 +1,147 @@ +using System.Collections.Concurrent; + +namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +/// +/// Tracks which sessions currently have at least one live +/// subscriber, so can skip the redaction +/// clone and the group send for sessions nobody is watching. +/// +/// +/// SignalR does not expose group membership, so the hub mirrors its own +/// AddToGroup/RemoveFromGroup calls here. In the steady state no +/// browser is on a session-details page, yet every session's dashboard-mirror +/// subscriber still called Publish for every event — a deep protobuf +/// clone (values are redacted by default) plus a send to an empty group, per +/// event, thrown away. This registry is the cheap gate in front of that work. +/// +/// Per-connection subscriptions are tracked as well, because a browser tab that +/// simply goes away never calls UnsubscribeSession; the hub's +/// OnDisconnectedAsync releases everything the connection held. +/// +/// +public sealed class EventsHubViewerRegistry +{ + private readonly ConcurrentDictionary _viewersBySession = new(StringComparer.Ordinal); + + private readonly ConcurrentDictionary> _sessionsByConnection = + new(StringComparer.Ordinal); + + /// + /// Records that is watching + /// . Repeat calls for the same pair are + /// idempotent, so one always clears them. + /// + /// SignalR connection id of the subscriber. + /// Session id being watched. + public void AddViewer(string connectionId, string sessionId) + { + if (string.IsNullOrWhiteSpace(connectionId) || string.IsNullOrWhiteSpace(sessionId)) + { + return; + } + + ConcurrentDictionary sessions = _sessionsByConnection.GetOrAdd( + connectionId, + static _ => new ConcurrentDictionary(StringComparer.Ordinal)); + + // The per-connection set is the source of truth for the count: only a + // subscription that was genuinely new increments the session's viewers. + if (!sessions.TryAdd(sessionId, 0)) + { + return; + } + + _viewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1); + } + + /// + /// Records that stopped watching + /// . A removal with no matching + /// is a no-op, so the count cannot go negative. + /// + /// SignalR connection id of the subscriber. + /// Session id no longer being watched. + public void RemoveViewer(string connectionId, string sessionId) + { + if (string.IsNullOrWhiteSpace(connectionId) || string.IsNullOrWhiteSpace(sessionId)) + { + return; + } + + if (!_sessionsByConnection.TryGetValue(connectionId, out ConcurrentDictionary? sessions) + || !sessions.TryRemove(sessionId, out _)) + { + return; + } + + ReleaseSession(sessionId); + } + + /// + /// Releases every subscription held by . + /// Called from the hub's disconnect callback, which is the only reliable + /// signal for a browser tab that closed without unsubscribing. + /// + /// SignalR connection id that dropped. + public void ReleaseConnection(string connectionId) + { + if (string.IsNullOrWhiteSpace(connectionId)) + { + return; + } + + if (!_sessionsByConnection.TryRemove(connectionId, out ConcurrentDictionary? sessions)) + { + return; + } + + foreach (string sessionId in sessions.Keys) + { + // TryRemove, not a bare enumeration: a concurrent RemoveViewer on the + // same detached set must not let the session be decremented twice. + if (sessions.TryRemove(sessionId, out _)) + { + ReleaseSession(sessionId); + } + } + } + + /// Gets a value indicating whether any hub connection is watching the session. + /// Session id to test. + /// when at least one connection is subscribed. + public bool HasViewers(string sessionId) => + !string.IsNullOrEmpty(sessionId) + && _viewersBySession.TryGetValue(sessionId, out int count) + && count > 0; + + /// + /// 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. + /// + /// Session id whose count is released. + private void ReleaseSession(string sessionId) + { + while (true) + { + if (!_viewersBySession.TryGetValue(sessionId, out int count)) + { + return; + } + + if (count <= 1) + { + if (_viewersBySession.TryRemove(new KeyValuePair(sessionId, count))) + { + return; + } + } + else if (_viewersBySession.TryUpdate(sessionId, count - 1, count)) + { + return; + } + } + } +} 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 7ec3c86..701573c 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs @@ -11,16 +11,64 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard; /// Verifies that honours /// MxGateway:Dashboard:ShowTagValues (SEC-25): 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. +/// shared source event is never mutated. Also verifies the viewer gate — the +/// mirror does no work at all for a session nobody is watching. /// public sealed class DashboardEventBroadcasterTests { + /// An unwatched session costs neither a redaction clone nor a send. + [Fact] + public void Publish_WithNoRegisteredViewers_DoesNotCloneOrSend() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + MxEvent source = BuildEventWithValue(); + + broadcaster.Publish("session-1", source); + + Assert.Equal(0, hubContext.SendCount); + Assert.Null(hubContext.LastArgument); + } + + /// A viewer on a different session does not open the gate for this one. + [Fact] + public void Publish_WithViewersOnAnotherSessionOnly_DoesNotSend() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + viewers.AddViewer("conn-1", "session-2"); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(0, hubContext.SendCount); + } + + /// Once the last viewer leaves, the mirror stops sending again. + [Fact] + public void Publish_AfterLastViewerLeaves_StopsSending() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + viewers.AddViewer("conn-1", "session-1"); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + + broadcaster.Publish("session-1", BuildEventWithValue()); + Assert.Equal(1, hubContext.SendCount); + + viewers.RemoveViewer("conn-1", "session-1"); + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(1, hubContext.SendCount); + } + /// Values are stripped from the mirror when ShowTagValues is off; metadata survives. [Fact] public void Publish_WhenShowTagValuesFalse_RedactsValuesButKeepsMetadata() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); @@ -43,7 +91,7 @@ public sealed class DashboardEventBroadcasterTests public void Publish_WhenShowTagValuesFalse_DoesNotMutateSourceEvent() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); @@ -61,7 +109,7 @@ public sealed class DashboardEventBroadcasterTests public void Publish_WhenShowTagValuesTrue_KeepsValues() { CapturingHubContext hubContext = new(); - DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, WatchedSession1()); MxEvent source = BuildEventWithValue(); broadcaster.Publish("session-1", source); @@ -73,7 +121,10 @@ public sealed class DashboardEventBroadcasterTests Assert.NotNull(sent.OnAlarmTransition.LimitValue); } - private static DashboardEventBroadcaster Create(CapturingHubContext hubContext, bool showTagValues) + private static DashboardEventBroadcaster Create( + CapturingHubContext hubContext, + bool showTagValues, + EventsHubViewerRegistry viewers) { GatewayOptions gatewayOptions = new() { @@ -82,10 +133,20 @@ public sealed class DashboardEventBroadcasterTests return new DashboardEventBroadcaster( hubContext, + viewers, Options.Create(gatewayOptions), NullLogger.Instance); } + /// A registry with one hub connection watching session-1. + /// The populated registry. + private static EventsHubViewerRegistry WatchedSession1() + { + EventsHubViewerRegistry viewers = new(); + viewers.AddViewer("conn-1", "session-1"); + return viewers; + } + private static MxEvent BuildEventWithValue() { return new MxEvent @@ -117,6 +178,9 @@ public sealed class DashboardEventBroadcasterTests /// Gets the first argument of the most recent send call. public object? LastArgument => _clients.GroupProxy.LastArgument; + + /// Gets the number of send calls this fake has observed. + public int SendCount => _clients.GroupProxy.SendCount; } private sealed class CapturingHubClients : IHubClients @@ -148,6 +212,9 @@ public sealed class DashboardEventBroadcasterTests /// Gets the first argument of the most recent send call. public object? LastArgument { get; private set; } + /// Gets the number of send calls made through this proxy. + public int SendCount { get; private set; } + /// Records the send call arguments and completes synchronously. /// The SignalR method name. /// The method arguments. @@ -155,6 +222,7 @@ public sealed class DashboardEventBroadcasterTests /// A completed task. public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default) { + SendCount++; LastArgument = args.Length > 0 ? args[0] : null; return Task.CompletedTask; } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs new file mode 100644 index 0000000..e220b9c --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs @@ -0,0 +1,165 @@ +using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard; + +/// +/// Verifies , the seam +/// consults before cloning and sending +/// an event: a session is "watched" only while at least one hub connection +/// holds a subscription to it, and a dropped connection releases every +/// subscription it held. +/// +public sealed class EventsHubViewerRegistryTests +{ + /// A session with no subscriber is not watched. + [Fact] + public void HasViewers_WithNoSubscribers_IsFalse() + { + EventsHubViewerRegistry registry = new(); + + Assert.False(registry.HasViewers("session-1")); + } + + /// Adding then removing the only viewer flips the session back to unwatched. + [Fact] + public void AddViewer_ThenRemoveViewer_TogglesWatchedState() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + Assert.True(registry.HasViewers("session-1")); + + registry.RemoveViewer("conn-1", "session-1"); + Assert.False(registry.HasViewers("session-1")); + } + + /// Each connection counts once: the session stays watched until the last one leaves. + [Fact] + public void RemoveViewer_WithOtherConnectionsStillSubscribed_KeepsSessionWatched() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + registry.AddViewer("conn-2", "session-1"); + + registry.RemoveViewer("conn-1", "session-1"); + Assert.True(registry.HasViewers("session-1")); + + registry.RemoveViewer("conn-2", "session-1"); + Assert.False(registry.HasViewers("session-1")); + } + + /// A repeated subscribe from the same connection is idempotent, so one unsubscribe clears it. + [Fact] + public void AddViewer_CalledTwiceForSameConnection_CountsOnce() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + registry.AddViewer("conn-1", "session-1"); + + registry.RemoveViewer("conn-1", "session-1"); + + Assert.False(registry.HasViewers("session-1")); + } + + /// Viewer counts are tracked per session; unrelated sessions stay unwatched. + [Fact] + public void AddViewer_TracksSessionsIndependently() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + + Assert.True(registry.HasViewers("session-1")); + Assert.False(registry.HasViewers("session-2")); + } + + /// A dropped connection releases every session it held. + [Fact] + public void ReleaseConnection_ReleasesEverySessionTheConnectionHeld() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + registry.AddViewer("conn-1", "session-2"); + registry.AddViewer("conn-2", "session-2"); + + registry.ReleaseConnection("conn-1"); + + Assert.False(registry.HasViewers("session-1")); + + // conn-2 still watches session-2. + Assert.True(registry.HasViewers("session-2")); + } + + /// Releasing a connection twice does not double-decrement another connection's subscription. + [Fact] + public void ReleaseConnection_CalledTwice_DoesNotDropOtherViewers() + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer("conn-1", "session-1"); + registry.AddViewer("conn-2", "session-1"); + + registry.ReleaseConnection("conn-1"); + registry.ReleaseConnection("conn-1"); + + Assert.True(registry.HasViewers("session-1")); + } + + /// Unmatched removals cannot drive the count negative and strand a session as unwatched. + [Fact] + public void RemoveViewer_WithoutMatchingAdd_LeavesCountAtZero() + { + EventsHubViewerRegistry registry = new(); + + registry.RemoveViewer("conn-1", "session-1"); + registry.RemoveViewer("conn-1", "session-1"); + registry.ReleaseConnection("conn-1"); + + Assert.False(registry.HasViewers("session-1")); + + // A subsequent genuine subscribe must still register as exactly one viewer. + registry.AddViewer("conn-1", "session-1"); + Assert.True(registry.HasViewers("session-1")); + + registry.RemoveViewer("conn-1", "session-1"); + Assert.False(registry.HasViewers("session-1")); + } + + /// Blank connection or session ids are ignored rather than tracked. + [Theory] + [InlineData("", "session-1")] + [InlineData(" ", "session-1")] + [InlineData("conn-1", "")] + [InlineData("conn-1", " ")] + public void AddViewer_WithBlankIdentifiers_IsIgnored(string connectionId, string sessionId) + { + EventsHubViewerRegistry registry = new(); + + registry.AddViewer(connectionId, sessionId); + + Assert.False(registry.HasViewers(sessionId)); + Assert.False(registry.HasViewers("session-1")); + } + + /// Concurrent add/remove pairs settle at zero viewers, never at a stuck-on count. + [Fact] + public void AddAndRemoveViewer_UnderConcurrency_SettlesAtZero() + { + EventsHubViewerRegistry registry = new(); + + Parallel.For(0, 64, i => + { + string connectionId = $"conn-{i}"; + for (int pass = 0; pass < 50; pass++) + { + registry.AddViewer(connectionId, "session-1"); + registry.RemoveViewer(connectionId, "session-1"); + } + }); + + Assert.False(registry.HasViewers("session-1")); + } +}