From b0f5941e46d14a5382da302c8a74b2c43381ea98 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 20:32:07 -0400 Subject: [PATCH] fix(dashboard): generation-tagged feed subscribers survive pump teardown races; drain-timeout observability --- .../Dashboard/Components/DashboardPageBase.cs | 30 ++- .../Dashboard/DashboardSnapshotFeed.cs | 248 +++++++++++++----- .../Dashboard/DashboardSnapshotFeedTests.cs | 119 ++++++++- 3 files changed, 330 insertions(+), 67 deletions(-) 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 _pushes = Channel.CreateUnbounded(); + private readonly TaskCompletionSource _disposalReached = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _disposalRelease = new(TaskCreationOptions.RunContinuationsAsynchronously); private int _enumerationCount; private int _completedEnumerationCount; private volatile bool _lastEnumerationWasCancelled; + private volatile bool _holdDisposal; /// Gets the number of times the feed started enumerating this source. public int EnumerationCount => Volatile.Read(ref _enumerationCount); + /// Gets a task that completes when a held enumerator disposal is reached. + public Task DisposalReached => _disposalReached.Task; + /// Gets the number of enumerations that have finished (cancelled, faulted, or completed). public int CompletedEnumerationCount => Volatile.Read(ref _completedEnumerationCount); @@ -303,11 +362,34 @@ public sealed class DashboardSnapshotFeedTests /// Exception to throw from the enumeration. public void Fault(Exception error) => _pushes.Writer.TryWrite(error); + /// + /// Parks enumerator disposal until , which holds the pump + /// in the window between observing the source's failure and detaching its subscribers. + /// + public void HoldDisposal() => _holdDisposal = true; + + /// Releases a held enumerator disposal. + public void ReleaseDisposal() => _disposalRelease.TrySetResult(); + /// public DashboardSnapshot GetSnapshot() => CreateSnapshot("current"); /// - public async IAsyncEnumerable WatchSnapshotsAsync( + public IAsyncEnumerable WatchSnapshotsAsync(CancellationToken cancellationToken) + => new GatedEnumerable(this); + + private async Task OnDisposingAsync() + { + if (!_holdDisposal) + { + return; + } + + _disposalReached.TrySetResult(); + await _disposalRelease.Task.ConfigureAwait(false); + } + + private async IAsyncEnumerable EnumerateAsync( [EnumeratorCancellation] CancellationToken cancellationToken) { Interlocked.Increment(ref _enumerationCount); @@ -329,5 +411,40 @@ public sealed class DashboardSnapshotFeedTests Interlocked.Increment(ref _completedEnumerationCount); } } + + /// + /// Wraps the iterator so disposal is a control point of its own: the feed ends a pump + /// generation when MoveNextAsync fails, which is strictly before this disposal runs. + /// + /// The fake whose enumeration is being wrapped. + private sealed class GatedEnumerable(FakeSnapshotService owner) : IAsyncEnumerable + { + /// Creates a gated enumerator over the fake's enumeration. + /// Token to observe for cancellation. + /// The gated enumerator. + public IAsyncEnumerator GetAsyncEnumerator( + CancellationToken cancellationToken = default) + => new GatedEnumerator(owner, owner.EnumerateAsync(cancellationToken).GetAsyncEnumerator(cancellationToken)); + } + + private sealed class GatedEnumerator( + FakeSnapshotService owner, + IAsyncEnumerator inner) : IAsyncEnumerator + { + /// Gets the current snapshot. + public DashboardSnapshot Current => inner.Current; + + /// Advances the wrapped enumeration. + /// A task that yields whether another snapshot is available. + public ValueTask MoveNextAsync() => inner.MoveNextAsync(); + + /// Parks while the fake holds disposal, then disposes the wrapped enumeration. + /// A task that represents the asynchronous operation. + public async ValueTask DisposeAsync() + { + await owner.OnDisposingAsync().ConfigureAwait(false); + await inner.DisposeAsync().ConfigureAwait(false); + } + } } }