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")); + } +}