diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs
index ded13c8..670b8fe 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs
@@ -1,80 +1,96 @@
using Microsoft.AspNetCore.Components;
-using Microsoft.AspNetCore.SignalR.Client;
-using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Components;
///
-/// Base class for Blazor dashboard pages that watch gateway metrics
-/// snapshots. The previous implementation polled
-/// directly; we
-/// now subscribe to so updates are
-/// pushed and disconnects survive reconnects via SignalR's
-/// auto-reconnect.
+/// Base class for Blazor dashboard pages that watch gateway metrics snapshots.
+/// Pages subscribe to the in-process , which
+/// multicasts a single
+/// enumeration to every circuit. An earlier implementation had each page open its
+/// own SignalR connection to /hubs/snapshot — a loopback WebSocket back into
+/// this same process, per page. The snapshot hub and its publisher remain for
+/// external (non-circuit) clients; server-rendered pages no longer use them.
///
public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
{
- private HubConnection? _hub;
+ ///
+ /// Upper bound on waiting for the watch loop while disposing. The loop marshals
+ /// renders through the renderer's dispatcher and disposal can run on that same
+ /// dispatcher, so the wait is bounded rather than unconditional.
+ ///
+ private static readonly TimeSpan WatchDrainTimeout = TimeSpan.FromSeconds(5);
- /// Snapshot service used to seed the initial render before the hub connects.
+ private readonly CancellationTokenSource _watchCancellation = new();
+ private Task? _watchTask;
+
+ /// Snapshot service used to seed the initial render before the first feed update.
[Inject]
protected IDashboardSnapshotService SnapshotService { get; set; } = null!;
- /// Factory that builds the SignalR connection (mints the hub bearer token).
+ /// Shared in-process snapshot feed this page renders from.
[Inject]
- protected DashboardHubConnectionFactory HubFactory { get; set; } = null!;
+ protected IDashboardSnapshotFeed SnapshotFeed { get; set; } = null!;
///
/// The most recent gateway metric snapshot. Synchronously seeded from
- /// for the very
- /// first render, then refreshed by hub push.
+ /// for the very first
+ /// render, then refreshed from the feed.
///
protected DashboardSnapshot? Snapshot { get; private set; }
///
- protected override async Task OnInitializedAsync()
+ protected override Task OnInitializedAsync()
{
Snapshot = SnapshotService.GetSnapshot();
- await ConnectHubAsync().ConfigureAwait(false);
+
+ // Deliberately not awaited: the watch loop runs for the lifetime of the page
+ // and is cancelled and drained by DisposeAsync.
+ _watchTask = WatchSnapshotsAsync(_watchCancellation.Token);
+ return Task.CompletedTask;
}
- /// Disposes the SignalR hub connection created for this page, tolerating disposal-time errors.
+ /// Cancels the snapshot subscription created for this page, tolerating disposal-time errors.
/// A task that represents the asynchronous operation.
public async ValueTask DisposeAsync()
{
- if (_hub is not null)
- {
- try
- {
- await _hub.DisposeAsync().ConfigureAwait(false);
- }
- catch
- {
- // Disposal-time errors are best-effort.
- }
- }
-
- GC.SuppressFinalize(this);
- }
-
- private async Task ConnectHubAsync()
- {
- _hub = HubFactory.Create("/hubs/snapshot");
- _hub.On(DashboardSnapshotHub.SnapshotMessage, async snapshot =>
- {
- Snapshot = snapshot;
- await InvokeAsync(StateHasChanged).ConfigureAwait(false);
- });
-
try
{
- await _hub.StartAsync().ConfigureAwait(false);
+ await _watchCancellation.CancelAsync().ConfigureAwait(false);
+
+ if (_watchTask is not null)
+ {
+ await _watchTask.WaitAsync(WatchDrainTimeout).ConfigureAwait(false);
+ }
}
catch
{
- // Hub is best-effort; the initial GetSnapshot() seed remains
- // valid and the snapshot service keeps populating its cache for
- // the next reconnect cycle.
+ // Disposal-time errors (including a drain timeout) are best-effort.
+ }
+
+ _watchCancellation.Dispose();
+ GC.SuppressFinalize(this);
+ }
+
+ private async Task WatchSnapshotsAsync(CancellationToken cancellationToken)
+ {
+ try
+ {
+ await foreach (DashboardSnapshot snapshot in SnapshotFeed
+ .WatchAsync(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ Snapshot = snapshot;
+ await InvokeAsync(StateHasChanged).ConfigureAwait(false);
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // The page is going away.
+ }
+ catch
+ {
+ // The feed is best-effort: the last rendered snapshot stays on screen and
+ // the snapshot service keeps serving GetSnapshot() for the next page load.
}
}
}
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs
index 42e014f..aece649 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardServiceCollectionExtensions.cs
@@ -38,6 +38,7 @@ public static class DashboardServiceCollectionExtensions
services.AddZbLdapAuth(configuration, "MxGateway:Ldap");
services.AddSingleton();
+ services.AddSingleton();
services.AddSingleton();
services.AddSingleton();
services.AddSingleton, DashboardGroupRoleMapper>();
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs
new file mode 100644
index 0000000..72a89cb
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs
@@ -0,0 +1,216 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
+
+///
+/// Fans one enumeration out to
+/// every dashboard circuit. The underlying watch is not multicast — each enumeration owns a
+/// timer and builds its own snapshot per tick — so subscribing per page would multiply the
+/// snapshot cost by the number of open pages.
+///
+///
+/// The pump is idle-gated: it starts when the subscriber count goes 0 → 1 and is cancelled
+/// and awaited when it goes 1 → 0, so an unwatched gateway runs no timer and builds no
+/// snapshots. Successive pumps are chained through _pumpTask, so a rapid
+/// unsubscribe/resubscribe restarts a fresh pump without ever running two enumerations at
+/// once.
+///
+public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
+{
+ private readonly IDashboardSnapshotService _snapshotService;
+ private readonly ILogger _logger;
+ private readonly object _gate = new();
+ private readonly List> _subscribers = [];
+
+ ///
+ /// The most recent pump, completed while idle. A starting pump awaits its predecessor
+ /// before enumerating, which is what guarantees a single live enumeration.
+ ///
+ private Task _pumpTask = Task.CompletedTask;
+
+ /// Cancellation for the live pump; null when no pump is running or one is being torn down.
+ private CancellationTokenSource? _pumpCancellation;
+
+ /// Initializes a new instance of the class.
+ /// Snapshot source to multicast.
+ /// Optional logger for pump faults.
+ public DashboardSnapshotFeed(
+ IDashboardSnapshotService snapshotService,
+ ILogger? logger = null)
+ {
+ _snapshotService = snapshotService ?? throw new ArgumentNullException(nameof(snapshotService));
+ _logger = logger ?? NullLogger.Instance;
+ }
+
+ ///
+ public async IAsyncEnumerable WatchAsync(
+ [EnumeratorCancellation] CancellationToken cancellationToken)
+ {
+ // Capacity 1 + DropOldest: a viewer only ever wants the latest snapshot, so a
+ // circuit that renders slowly neither buffers without bound nor blocks the pump
+ // (TryWrite always succeeds) — it just skips the snapshots it was too slow for.
+ Channel channel = Channel.CreateBounded(
+ new BoundedChannelOptions(1)
+ {
+ FullMode = BoundedChannelFullMode.DropOldest,
+ SingleReader = true,
+ SingleWriter = false,
+ });
+
+ Subscribe(channel);
+ try
+ {
+ await foreach (DashboardSnapshot snapshot in channel.Reader
+ .ReadAllAsync(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ yield return snapshot;
+ }
+ }
+ finally
+ {
+ // Untokened on purpose: teardown must run to completion even when this
+ // subscriber is unwinding because its own token fired.
+ await UnsubscribeAsync(channel).ConfigureAwait(false);
+ }
+ }
+
+ private void Subscribe(Channel channel)
+ {
+ lock (_gate)
+ {
+ _subscribers.Add(channel);
+ if (_subscribers.Count != 1)
+ {
+ return;
+ }
+
+ CancellationTokenSource cancellation = new();
+ Task previous = _pumpTask;
+ _pumpCancellation = cancellation;
+
+ // Task.Run, not a direct call: an async iterator runs synchronously up to its
+ // first suspension, and the first pull of the underlying watch can read the API
+ // key table. That must not run on the subscribing circuit's thread, let alone
+ // while this lock is held.
+ _pumpTask = Task.Run(() => PumpAsync(previous, cancellation, cancellation.Token));
+ }
+ }
+
+ private async Task UnsubscribeAsync(Channel channel)
+ {
+ CancellationTokenSource? cancellation;
+ Task pump;
+ lock (_gate)
+ {
+ if (!_subscribers.Remove(channel) || _subscribers.Count != 0)
+ {
+ // Either the pump already dropped this channel (it completed or faulted
+ // and reset itself), or other viewers are still watching.
+ return;
+ }
+
+ cancellation = _pumpCancellation;
+ _pumpCancellation = null;
+ pump = _pumpTask;
+ }
+
+ try
+ {
+ cancellation?.Cancel();
+ }
+ catch (ObjectDisposedException)
+ {
+ // The pump reset itself and disposed its own cancellation source first.
+ }
+
+ try
+ {
+ await pump.ConfigureAwait(false);
+ }
+ catch (Exception)
+ {
+ // A pump fault has already been reported to the subscribers it had; the
+ // unsubscribing caller is only waiting for the enumeration to stop.
+ }
+ }
+
+ private async Task PumpAsync(Task previous, CancellationTokenSource cancellation, CancellationToken cancellationToken)
+ {
+ try
+ {
+ // Never overlap with the enumeration this pump replaces.
+ await previous.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+
+ await foreach (DashboardSnapshot snapshot in _snapshotService
+ .WatchSnapshotsAsync(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ Broadcast(snapshot);
+ }
+
+ // The source completed on its own; hand the completion to the subscribers
+ // and re-arm so the next one starts a fresh enumeration.
+ Reset(cancellation, error: null);
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ // Last subscriber left: the unsubscribing caller already detached the
+ // channels and cleared the pump state.
+ }
+ catch (Exception error)
+ {
+ _logger.LogWarning(error, "Dashboard snapshot feed stopped; the next subscriber restarts it.");
+ Reset(cancellation, error);
+ }
+ finally
+ {
+ cancellation.Dispose();
+ }
+ }
+
+ private void Broadcast(DashboardSnapshot snapshot)
+ {
+ lock (_gate)
+ {
+ foreach (Channel subscriber in _subscribers)
+ {
+ // Bounded/DropOldest: always accepted unless the channel is completed.
+ subscriber.Writer.TryWrite(snapshot);
+ }
+ }
+ }
+
+ ///
+ /// Detaches every subscriber and clears the pump state so the next subscriber starts a
+ /// new enumeration. The detached subscribers observe (or a
+ /// clean end of stream) from their own WatchAsync.
+ ///
+ /// The calling pump's cancellation source, used as its ownership token.
+ /// Failure to surface, or null when the source completed cleanly.
+ private void Reset(CancellationTokenSource cancellation, Exception? error)
+ {
+ Channel[] detached;
+ lock (_gate)
+ {
+ if (!ReferenceEquals(_pumpCancellation, cancellation))
+ {
+ // A newer pump (or an in-flight teardown) owns the state now: its
+ // subscribers must not be detached by this pump's exit.
+ return;
+ }
+
+ detached = _subscribers.ToArray();
+ _subscribers.Clear();
+ _pumpCancellation = null;
+ }
+
+ foreach (Channel subscriber in detached)
+ {
+ subscriber.Writer.TryComplete(error);
+ }
+ }
+}
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardSnapshotFeed.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardSnapshotFeed.cs
new file mode 100644
index 0000000..7e72875
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardSnapshotFeed.cs
@@ -0,0 +1,26 @@
+namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
+
+///
+/// In-process multicast over .
+/// One enumeration of the underlying watch is fanned out to every subscriber, so N
+/// dashboard circuits cost one snapshot build per tick instead of N — and while nobody
+/// subscribes, nothing runs at all.
+///
+///
+/// There is no authentication or authorization gate here: the feed is reached only from
+/// Blazor dashboard components, whose endpoints already require
+/// , so every caller is a circuit
+/// authorized as Viewer. Remote (non-circuit) consumers still go through
+/// /hubs/snapshot, which applies the hub authorization policy itself.
+///
+public interface IDashboardSnapshotFeed
+{
+ ///
+ /// Watches the shared snapshot stream. Each caller gets the snapshots produced while
+ /// it is subscribed; a caller that reads slowly sees only the newest snapshot rather
+ /// than a backlog, and never delays the other subscribers.
+ ///
+ /// Token that ends this caller's subscription.
+ /// An asynchronous stream of dashboard snapshots.
+ IAsyncEnumerable WatchAsync(CancellationToken cancellationToken);
+}
diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs
new file mode 100644
index 0000000..b86bfd7
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs
@@ -0,0 +1,333 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using ZB.MOM.WW.MxGateway.Server.Dashboard;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard;
+
+///
+/// Covers the in-process snapshot fan-out that replaced the dashboard pages'
+/// loopback /hubs/snapshot connections. The invariants under test are the
+/// ones that make the feed cheaper than the hub hop: exactly one underlying
+/// enumeration for any
+/// number of viewers, nothing at all while nobody is watching, and a slow viewer
+/// that can neither buffer without bound nor stall the others.
+///
+public sealed class DashboardSnapshotFeedTests
+{
+ private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
+
+ ///
+ /// With no page subscribed, the feed must not touch the snapshot service at
+ /// all — no timer, no snapshot build. This is the whole point of the idle
+ /// gate: an unattended gateway does no dashboard work.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WithNoSubscribers_NeverEnumeratesUnderlyingWatch()
+ {
+ FakeSnapshotService service = new();
+ DashboardSnapshotFeed feed = new(service);
+
+ // Obtaining the enumerable without enumerating it must not subscribe
+ // either: the pump starts on the first MoveNextAsync, not before.
+ _ = feed.WatchAsync(CancellationToken.None);
+ await Task.Delay(TimeSpan.FromMilliseconds(100));
+
+ Assert.Equal(0, service.EnumerationCount);
+ }
+
+ ///
+ /// Two viewers share one underlying enumeration and both see the same
+ /// pushed snapshot. Before the feed, each page opened its own SignalR
+ /// connection and the publisher pulled its own snapshot stream.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WithTwoSubscribers_SharesASingleUnderlyingEnumeration()
+ {
+ FakeSnapshotService service = new();
+ DashboardSnapshotFeed feed = new(service);
+
+ using CancellationTokenSource firstCancellation = new();
+ using CancellationTokenSource secondCancellation = new();
+ IAsyncEnumerator first =
+ feed.WatchAsync(firstCancellation.Token).GetAsyncEnumerator(firstCancellation.Token);
+ Task firstMove = first.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+ IAsyncEnumerator second =
+ feed.WatchAsync(secondCancellation.Token).GetAsyncEnumerator(secondCancellation.Token);
+ Task secondMove = second.MoveNextAsync().AsTask();
+
+ // Push until both have observed a snapshot: a subscriber only becomes
+ // visible to the pump once its MoveNextAsync has registered the channel,
+ // so a single push could race the second registration.
+ await PushUntilAsync(service, Task.WhenAll(firstMove, secondMove));
+
+ Assert.True(await firstMove.WaitAsync(TestTimeout));
+ Assert.True(await secondMove.WaitAsync(TestTimeout));
+ Assert.StartsWith("push-", first.Current.GatewayVersion, StringComparison.Ordinal);
+ Assert.StartsWith("push-", second.Current.GatewayVersion, StringComparison.Ordinal);
+ Assert.Equal(1, service.EnumerationCount);
+
+ await firstCancellation.CancelAsync();
+ await secondCancellation.CancelAsync();
+ await DrainAsync(first, firstMove);
+ await DrainAsync(second, secondMove);
+ }
+
+ ///
+ /// The last viewer leaving must cancel the underlying enumeration (idle
+ /// gate re-armed), and the next viewer must restart it — the rapid
+ /// unsubscribe/resubscribe path a page navigation exercises.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WhenLastSubscriberLeaves_CancelsPumpAndRestartsForTheNext()
+ {
+ FakeSnapshotService service = new();
+ DashboardSnapshotFeed feed = new(service);
+
+ using CancellationTokenSource firstCancellation = new();
+ IAsyncEnumerator first =
+ feed.WatchAsync(firstCancellation.Token).GetAsyncEnumerator(firstCancellation.Token);
+ Task firstMove = first.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+ await firstCancellation.CancelAsync();
+ await DrainAsync(first, firstMove);
+
+ await WaitUntilAsync(() => service.CompletedEnumerationCount >= 1);
+ Assert.True(service.LastEnumerationWasCancelled);
+
+ using CancellationTokenSource secondCancellation = new();
+ IAsyncEnumerator second =
+ feed.WatchAsync(secondCancellation.Token).GetAsyncEnumerator(secondCancellation.Token);
+ Task secondMove = second.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 2);
+
+ Assert.Equal(2, service.EnumerationCount);
+
+ await secondCancellation.CancelAsync();
+ await DrainAsync(second, secondMove);
+ }
+
+ ///
+ /// A viewer that is not reading must not stall the pump or accumulate
+ /// snapshots: its bounded channel drops the oldest, so its next read is the
+ /// newest snapshot the pump has broadcast, not a backlog head. The fast
+ /// reader's progress is what makes the assertion deterministic — once it has
+ /// seen the third snapshot the pump has provably broadcast all three.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WithASlowSubscriber_KeepsOnlyTheNewestSnapshot()
+ {
+ FakeSnapshotService service = new();
+ DashboardSnapshotFeed feed = new(service);
+
+ using CancellationTokenSource fastCancellation = new();
+ using CancellationTokenSource slowCancellation = new();
+ IAsyncEnumerator fast =
+ feed.WatchAsync(fastCancellation.Token).GetAsyncEnumerator(fastCancellation.Token);
+ Task fastMove = fast.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+ // The slow subscriber registers but never advances until the very end.
+ IAsyncEnumerator slow =
+ feed.WatchAsync(slowCancellation.Token).GetAsyncEnumerator(slowCancellation.Token);
+ Task slowMove = slow.MoveNextAsync().AsTask();
+ await PushUntilAsync(service, Task.WhenAll(fastMove, slowMove));
+ Assert.True(await fastMove.WaitAsync(TestTimeout));
+ Assert.True(await slowMove.WaitAsync(TestTimeout));
+
+ service.Push(CreateSnapshot("s1"));
+ service.Push(CreateSnapshot("s2"));
+ service.Push(CreateSnapshot("s3"));
+
+ // Drain the fast reader until it sees s3; that proves the pump broadcast
+ // all three to every subscriber, so the slow channel now holds exactly s3.
+ string fastLatest = fast.Current.GatewayVersion;
+ while (fastLatest != "s3")
+ {
+ Assert.True(await fast.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+ fastLatest = fast.Current.GatewayVersion;
+ }
+
+ Assert.True(await slow.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+ Assert.Equal("s3", slow.Current.GatewayVersion);
+
+ await fastCancellation.CancelAsync();
+ await slowCancellation.CancelAsync();
+ await DrainAsync(fast, Task.FromResult(true));
+ await DrainAsync(slow, Task.FromResult(true));
+ }
+
+ ///
+ /// A fault in the underlying watch is surfaced to the current viewers rather
+ /// than silently hanging them, and it resets the feed so the next viewer
+ /// starts a fresh pump instead of attaching to a dead one.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WhenUnderlyingWatchFaults_PropagatesAndRestartsForTheNextSubscriber()
+ {
+ FakeSnapshotService service = new();
+ DashboardSnapshotFeed feed = new(service);
+
+ using CancellationTokenSource firstCancellation = new();
+ IAsyncEnumerator first =
+ feed.WatchAsync(firstCancellation.Token).GetAsyncEnumerator(firstCancellation.Token);
+ Task firstMove = first.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 1);
+
+ service.Fault(new InvalidOperationException("simulated snapshot source failure"));
+
+ InvalidOperationException failure =
+ await Assert.ThrowsAsync(() => firstMove.WaitAsync(TestTimeout));
+ Assert.Equal("simulated snapshot source failure", failure.Message);
+ await first.DisposeAsync();
+
+ using CancellationTokenSource secondCancellation = new();
+ IAsyncEnumerator second =
+ feed.WatchAsync(secondCancellation.Token).GetAsyncEnumerator(secondCancellation.Token);
+ Task secondMove = second.MoveNextAsync().AsTask();
+ await WaitUntilAsync(() => service.EnumerationCount >= 2);
+
+ await PushUntilAsync(service, secondMove);
+ Assert.True(await secondMove.WaitAsync(TestTimeout));
+
+ await secondCancellation.CancelAsync();
+ await DrainAsync(second, secondMove);
+ }
+
+ /// Builds a snapshot whose version string identifies it in assertions.
+ /// Identity marker carried in GatewayVersion.
+ /// A snapshot carrying the supplied identity marker.
+ private static DashboardSnapshot CreateSnapshot(string version)
+ {
+ return new DashboardSnapshot(
+ GeneratedAt: DateTimeOffset.UnixEpoch,
+ GatewayStartedAt: DateTimeOffset.UnixEpoch,
+ GatewayUptime: TimeSpan.Zero,
+ GatewayStatus: "Healthy",
+ GatewayVersion: version,
+ Sessions: Array.Empty(),
+ Workers: Array.Empty(),
+ Metrics: Array.Empty(),
+ Faults: Array.Empty(),
+ ApiKeys: Array.Empty(),
+ Configuration: null!,
+ Galaxy: null!);
+ }
+
+ ///
+ /// Pushes snapshots until the supplied task completes, so a test never
+ /// depends on a single push landing after a subscriber has registered.
+ ///
+ /// Fake snapshot source to push through.
+ /// Task whose completion stops the pushes.
+ /// A task that represents the asynchronous operation.
+ private static async Task PushUntilAsync(FakeSnapshotService service, Task until)
+ {
+ using CancellationTokenSource cancellation = new(TestTimeout);
+ int sequence = 0;
+ while (!until.IsCompleted)
+ {
+ service.Push(CreateSnapshot($"push-{sequence++}"));
+ await Task.Delay(TimeSpan.FromMilliseconds(5), cancellation.Token);
+ }
+ }
+
+ ///
+ /// Observes the cancellation of a pending enumeration and disposes the
+ /// enumerator, mirroring how await foreach unwinds a cancelled watch.
+ ///
+ /// Enumerator to unwind.
+ /// The in-flight move, if any.
+ /// A task that represents the asynchronous operation.
+ private static async Task DrainAsync(IAsyncEnumerator enumerator, Task pending)
+ {
+ try
+ {
+ await pending.WaitAsync(TestTimeout);
+ }
+ catch (OperationCanceledException)
+ {
+ }
+
+ try
+ {
+ await enumerator.DisposeAsync();
+ }
+ catch (OperationCanceledException)
+ {
+ }
+ }
+
+ private static async Task WaitUntilAsync(Func predicate)
+ {
+ using CancellationTokenSource cancellation = new(TestTimeout);
+ while (!predicate())
+ {
+ await Task.Delay(TimeSpan.FromMilliseconds(5), cancellation.Token);
+ }
+ }
+
+ ///
+ /// Snapshot source under the feed's control: counts enumerations, records how
+ /// each one ended, and lets the test drive snapshots (or a fault) into the
+ /// live enumeration.
+ ///
+ private sealed class FakeSnapshotService : IDashboardSnapshotService
+ {
+ private readonly Channel