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 670b8fe..ca13526 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/DashboardPageBase.cs
@@ -1,4 +1,5 @@
using Microsoft.AspNetCore.Components;
+using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Components;
@@ -31,6 +32,10 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
[Inject]
protected IDashboardSnapshotFeed SnapshotFeed { get; set; } = null!;
+ /// Logger used to report a snapshot subscription that ended or would not drain.
+ [Inject]
+ protected ILogger? Logger { get; set; }
+
///
/// The most recent gateway metric snapshot. Synchronously seeded from
/// for the very first
@@ -62,9 +67,21 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
await _watchTask.WaitAsync(WatchDrainTimeout).ConfigureAwait(false);
}
}
+ catch (TimeoutException)
+ {
+ // Accepted limitation: the abandoned loop still holds its feed subscription, so
+ // the feed's idle gate stays open until it does unwind. There is no way to force
+ // a detach — the loop is parked on a dispatcher that is not draining — so the
+ // warning is the operator's only signal that a circuit teardown wedged.
+ Logger?.LogWarning(
+ "Dashboard page {Page} did not release its snapshot subscription within {Timeout}; "
+ + "the shared snapshot feed stays active until it unwinds.",
+ GetType().Name,
+ WatchDrainTimeout);
+ }
catch
{
- // Disposal-time errors (including a drain timeout) are best-effort.
+ // Other disposal-time errors are best-effort.
}
_watchCancellation.Dispose();
@@ -87,10 +104,15 @@ public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
{
// The page is going away.
}
- catch
+ catch (Exception error)
{
- // The feed is best-effort: the last rendered snapshot stays on screen and
- // the snapshot service keeps serving GetSnapshot() for the next page load.
+ // The feed is best-effort: the last rendered snapshot stays on screen and the
+ // snapshot service keeps serving GetSnapshot() for the next page load. Logged
+ // once here, on the way out of the loop — never per snapshot.
+ Logger?.LogWarning(
+ error,
+ "Live snapshot updates ended for dashboard page {Page}; it keeps the last rendered snapshot.",
+ GetType().Name);
}
}
}
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs
index 72a89cb..ede8f0e 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs
@@ -12,18 +12,30 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
/// 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
+///
+/// The pump is idle-gated: it starts when the first subscriber arrives and is cancelled and
+/// awaited when the last one leaves, 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.
+///
+///
+/// Every subscriber is tagged with the pump generation it joined under, and a dying pump only
+/// ever detaches its own generation. A pump ends its generation the instant its source fails
+/// or completes — before the (possibly slow) enumerator disposal — so a subscriber arriving
+/// while a pump unwinds starts a fresh generation instead of silently attaching to a dead
+/// pump that is about to detach everybody and leave nobody watching.
+///
///
public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
{
+ /// Generation value meaning "no pump is accepting subscribers".
+ private const long NoGeneration = 0;
+
private readonly IDashboardSnapshotService _snapshotService;
private readonly ILogger _logger;
private readonly object _gate = new();
- private readonly List> _subscribers = [];
+ private readonly List _subscribers = [];
///
/// The most recent pump, completed while idle. A starting pump awaits its predecessor
@@ -31,9 +43,15 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
///
private Task _pumpTask = Task.CompletedTask;
- /// Cancellation for the live pump; null when no pump is running or one is being torn down.
+ /// Cancellation for the live pump; null when no generation is accepting subscribers.
private CancellationTokenSource? _pumpCancellation;
+ /// The generation new subscribers join, or when no pump is live.
+ private long _generation = NoGeneration;
+
+ /// Last generation handed out; only ever incremented under _gate.
+ private long _lastGeneration = NoGeneration;
+
/// Initializes a new instance of the class.
/// Snapshot source to multicast.
/// Optional logger for pump faults.
@@ -60,7 +78,7 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
SingleWriter = false,
});
- Subscribe(channel);
+ Subscription subscription = Subscribe(channel);
try
{
await foreach (DashboardSnapshot snapshot in channel.Reader
@@ -74,47 +92,40 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
{
// Untokened on purpose: teardown must run to completion even when this
// subscriber is unwinding because its own token fired.
- await UnsubscribeAsync(channel).ConfigureAwait(false);
+ await UnsubscribeAsync(subscription).ConfigureAwait(false);
}
}
- private void Subscribe(Channel channel)
+ private Subscription 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));
+ // A live generation is joined; otherwise this subscriber starts one. Keying on
+ // "is a generation live" rather than "is this the first subscriber" is what makes
+ // a subscriber arriving while a pump unwinds start a fresh pump for itself.
+ long generation = _pumpCancellation is null ? StartPumpLocked() : _generation;
+ Subscription subscription = new(channel, generation);
+ _subscribers.Add(subscription);
+ return subscription;
}
}
- private async Task UnsubscribeAsync(Channel channel)
+ private async Task UnsubscribeAsync(Subscription subscription)
{
CancellationTokenSource? cancellation;
Task pump;
lock (_gate)
{
- if (!_subscribers.Remove(channel) || _subscribers.Count != 0)
+ if (!_subscribers.Remove(subscription) || _subscribers.Count != 0)
{
- // Either the pump already dropped this channel (it completed or faulted
- // and reset itself), or other viewers are still watching.
+ // Either the pump already detached this subscription (it completed or
+ // faulted), or other viewers are still watching.
return;
}
cancellation = _pumpCancellation;
_pumpCancellation = null;
+ _generation = NoGeneration;
pump = _pumpTask;
}
@@ -124,7 +135,7 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
}
catch (ObjectDisposedException)
{
- // The pump reset itself and disposed its own cancellation source first.
+ // The pump ended on its own and disposed its cancellation source first.
}
try
@@ -138,33 +149,90 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
}
}
- private async Task PumpAsync(Task previous, CancellationTokenSource cancellation, CancellationToken cancellationToken)
+ ///
+ /// Starts a pump generation. Must be called while holding _gate; the caller adds
+ /// the subscribers that belong to the returned generation.
+ ///
+ /// The new generation identifier.
+ private long StartPumpLocked()
+ {
+ long generation = ++_lastGeneration;
+ CancellationTokenSource cancellation = new();
+ Task previous = _pumpTask;
+ _generation = generation;
+ _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(generation, previous, cancellation, cancellation.Token));
+ return generation;
+ }
+
+ private async Task PumpAsync(
+ long generation,
+ 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
+ // Enumerated by hand rather than with await foreach so the generation can be
+ // ended the moment the source fails or completes — await foreach would run the
+ // enumerator's disposal first, and a subscriber arriving during that disposal
+ // would join a generation that is already doomed.
+ IAsyncEnumerator snapshots = _snapshotService
.WatchSnapshotsAsync(cancellationToken)
- .ConfigureAwait(false))
+ .GetAsyncEnumerator(cancellationToken);
+ try
{
- Broadcast(snapshot);
+ while (true)
+ {
+ bool moved;
+ try
+ {
+ moved = await snapshots.MoveNextAsync().ConfigureAwait(false);
+ }
+ catch
+ {
+ EndGeneration(generation);
+ throw;
+ }
+
+ if (!moved)
+ {
+ EndGeneration(generation);
+ break;
+ }
+
+ Broadcast(generation, snapshots.Current);
+ }
+ }
+ finally
+ {
+ await snapshots.DisposeAsync().ConfigureAwait(false);
}
- // 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);
+ // The source completed on its own; hand the completion to this generation's
+ // subscribers and re-arm so the next one starts a fresh enumeration.
+ Reset(generation, error: null);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
- // Last subscriber left: the unsubscribing caller already detached the
- // channels and cleared the pump state.
+ // The production DashboardSnapshotService swallows cancellation and yield-breaks,
+ // so normal teardown exits through the fall-through above (with an ownership-checked
+ // Reset that finds no subscribers); an implementation that propagates the token
+ // instead exits here. Both shapes end the generation exactly once.
+ EndGeneration(generation);
}
catch (Exception error)
{
_logger.LogWarning(error, "Dashboard snapshot feed stopped; the next subscriber restarts it.");
- Reset(cancellation, error);
+ Reset(generation, error);
}
finally
{
@@ -172,45 +240,101 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
}
}
- private void Broadcast(DashboardSnapshot snapshot)
+ private void Broadcast(long generation, DashboardSnapshot snapshot)
{
lock (_gate)
{
- foreach (Channel subscriber in _subscribers)
+ foreach (Subscription subscriber in _subscribers)
{
+ if (subscriber.Generation != generation)
+ {
+ continue;
+ }
+
// Bounded/DropOldest: always accepted unless the channel is completed.
- subscriber.Writer.TryWrite(snapshot);
+ subscriber.Channel.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.
+ /// Stops handing to new subscribers. Called the instant a
+ /// pump's source fails or completes, before its enumerator is disposed.
///
- /// 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)
+ /// The generation that has ended.
+ private void EndGeneration(long generation)
{
- 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);
+ EndGenerationLocked(generation);
}
}
+
+ /// Clears the live-pump state if still owns it.
+ /// The generation that has ended.
+ private void EndGenerationLocked(long generation)
+ {
+ if (_generation != generation)
+ {
+ return;
+ }
+
+ _generation = NoGeneration;
+ _pumpCancellation = null;
+ }
+
+ ///
+ /// Detaches the subscribers of a finished generation and re-arms the feed. Subscribers of
+ /// any other generation are left alone — they belong to a pump that is still running (or
+ /// about to), so a dying pump must not take them down with it.
+ ///
+ /// The generation whose subscribers are being detached.
+ /// Failure to surface, or null when the source completed cleanly.
+ private void Reset(long generation, Exception? error)
+ {
+ List> detached = [];
+ lock (_gate)
+ {
+ for (int index = _subscribers.Count - 1; index >= 0; index--)
+ {
+ if (_subscribers[index].Generation != generation)
+ {
+ continue;
+ }
+
+ detached.Add(_subscribers[index].Channel);
+ _subscribers.RemoveAt(index);
+ }
+
+ EndGenerationLocked(generation);
+
+ if (_subscribers.Count > 0 && _pumpCancellation is null)
+ {
+ // Belt and braces: subscribers left with no live pump would be frozen for
+ // good, because only a subscriber that finds no generation starts one.
+ long restarted = StartPumpLocked();
+ foreach (Subscription subscriber in _subscribers)
+ {
+ subscriber.Generation = restarted;
+ }
+ }
+ }
+
+ foreach (Channel channel in detached)
+ {
+ channel.Writer.TryComplete(error);
+ }
+ }
+
+ /// One viewer's delivery channel plus the pump generation serving it.
+ /// Delivery channel for this viewer.
+ /// Pump generation this viewer joined under.
+ private sealed class Subscription(Channel channel, long generation)
+ {
+ /// Gets the viewer's delivery channel.
+ public Channel Channel { get; } = channel;
+
+ /// Gets or sets the pump generation currently serving this viewer.
+ public long Generation { get; set; } = generation;
+ }
}
diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs
index b86bfd7..330ac17 100644
--- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs
@@ -201,6 +201,59 @@ public sealed class DashboardSnapshotFeedTests
await DrainAsync(second, secondMove);
}
+ ///
+ /// The race the generation tagging exists for: a page subscribes in the window between
+ /// the source failing and the dying pump detaching its subscribers. Without generations
+ /// the newcomer joined the doomed pump, was detached with its error, and no pump ever
+ /// restarted (only a first subscriber started one) — that page froze for good.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WatchAsync_WhenASubscriberJoinsWhileAFaultedPumpUnwinds_IsServedByAFreshPump()
+ {
+ FakeSnapshotService service = new();
+ service.HoldDisposal();
+ 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"));
+
+ // The pump has observed the failure and is parked disposing the enumerator — the
+ // exact window in which a page used to attach itself to a doomed pump.
+ await service.DisposalReached.WaitAsync(TestTimeout);
+
+ using CancellationTokenSource secondCancellation = new();
+ IAsyncEnumerator second =
+ feed.WatchAsync(secondCancellation.Token).GetAsyncEnumerator(secondCancellation.Token);
+
+ // An async iterator body runs synchronously up to its first await, so the
+ // subscription is registered by the time MoveNextAsync hands back its task.
+ Task secondMove = second.MoveNextAsync().AsTask();
+
+ service.ReleaseDisposal();
+
+ // The subscriber that was there when the source broke still learns about it...
+ InvalidOperationException failure =
+ await Assert.ThrowsAsync(() => firstMove.WaitAsync(TestTimeout));
+ Assert.Equal("simulated snapshot source failure", failure.Message);
+ await first.DisposeAsync();
+
+ // ...and the one that joined mid-unwind is served by a restarted enumeration
+ // instead of inheriting the failure.
+ await WaitUntilAsync(() => service.EnumerationCount >= 2);
+ await PushUntilAsync(service, secondMove);
+ Assert.True(await secondMove.WaitAsync(TestTimeout));
+ Assert.StartsWith("push-", second.Current.GatewayVersion, StringComparison.Ordinal);
+
+ 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.
@@ -282,13 +335,19 @@ public sealed class DashboardSnapshotFeedTests
private sealed class FakeSnapshotService : IDashboardSnapshotService
{
private readonly Channel