fix(dashboard): generation-scoped idle gate in UnsubscribeAsync — no zero-subscriber pump survives

This commit is contained in:
Joseph Doherty
2026-08-15 20:43:17 -04:00
parent b0f5941e46
commit 756296886b
2 changed files with 109 additions and 5 deletions
@@ -14,8 +14,9 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
/// <remarks>
/// <para>
/// 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 <c>_pumpTask</c>, so a rapid
/// awaited when the last subscriber <em>of the live generation</em> leaves, so an unwatched
/// gateway runs no timer and builds no snapshots. Successive pumps are chained through
/// <c>_pumpTask</c>, so a rapid
/// unsubscribe/resubscribe restarts a fresh pump without ever running two enumerations at
/// once.
/// </para>
@@ -116,10 +117,25 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
Task pump;
lock (_gate)
{
if (!_subscribers.Remove(subscription) || _subscribers.Count != 0)
if (!_subscribers.Remove(subscription))
{
// Either the pump already detached this subscription (it completed or
// faulted), or other viewers are still watching.
// The pump already detached this subscription (it completed or faulted).
return;
}
if (subscription.Generation != _generation)
{
// This viewer belonged to a generation that has already ended. The live
// pump — if there is one — serves other viewers and must not be cancelled
// on their behalf; the dying pump is stopping under its own steam.
return;
}
if (HasSubscribersLocked(_generation))
{
// Other viewers are still watching the live generation. Counting the whole
// list here would be wrong: subscribers of an ending generation linger in it
// until that pump's Reset runs, and they must not hold the idle gate open.
return;
}
@@ -149,6 +165,22 @@ public sealed class DashboardSnapshotFeed : IDashboardSnapshotFeed
}
}
/// <summary>Reports whether any subscriber is still being served by a generation.</summary>
/// <param name="generation">The generation to look for.</param>
/// <returns>True when at least one subscriber carries that generation.</returns>
private bool HasSubscribersLocked(long generation)
{
foreach (Subscription subscriber in _subscribers)
{
if (subscriber.Generation == generation)
{
return true;
}
}
return false;
}
/// <summary>
/// Starts a pump generation. Must be called while holding <c>_gate</c>; the caller adds
/// the subscribers that belong to the returned generation.
@@ -254,6 +254,72 @@ public sealed class DashboardSnapshotFeedTests
await DrainAsync(second, secondMove);
}
/// <summary>
/// The idle gate is per generation, not per subscriber count. A viewer that joins while a
/// pump unwinds starts a new generation, and the old generation's viewers linger in the
/// list until that pump's reset runs — so a global "is the list empty" check let the new
/// viewer leave without cancelling the generation it had just started, leaving a pump
/// enumerating the snapshot source with nobody watching it.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WatchAsync_WhenAJoinerLeavesWhileAnOldGenerationLingers_LeavesNoPumpRunning()
{
FakeSnapshotService service = new();
service.HoldDisposal();
DashboardSnapshotFeed feed = new(service);
using CancellationTokenSource oldCancellation = new();
IAsyncEnumerator<DashboardSnapshot> oldSubscriber =
feed.WatchAsync(oldCancellation.Token).GetAsyncEnumerator(oldCancellation.Token);
Task<bool> oldMove = oldSubscriber.MoveNextAsync().AsTask();
await WaitUntilAsync(() => service.EnumerationCount >= 1);
service.Fault(new InvalidOperationException("simulated snapshot source failure"));
await service.DisposalReached.WaitAsync(TestTimeout);
// Joins mid-unwind (starting a fresh generation) and leaves again before the dying
// pump has detached the subscriber that is still lingering in the list.
using CancellationTokenSource joinerCancellation = new();
IAsyncEnumerator<DashboardSnapshot> joiner =
feed.WatchAsync(joinerCancellation.Token).GetAsyncEnumerator(joinerCancellation.Token);
Task<bool> joinerMove = joiner.MoveNextAsync().AsTask();
await joinerCancellation.CancelAsync();
// The joiner's unwind is a continuation of its cancelled channel read; give it time to
// run its unsubscribe before the dying pump is released. Only the interleaving depends
// on this delay — the assertions below hold either way.
await Task.Delay(TimeSpan.FromMilliseconds(150));
service.ReleaseDisposal();
await Assert.ThrowsAsync<InvalidOperationException>(() => oldMove.WaitAsync(TestTimeout));
await oldSubscriber.DisposeAsync();
// Completes only once the joiner's unsubscribe has awaited its generation's pump.
await DrainAsync(joiner, joinerMove);
// Nobody is watching, so nothing may consume a snapshot: a surviving pump would drain
// this push within its first read.
service.Push(CreateSnapshot("orphan-check"));
await Task.Delay(TimeSpan.FromMilliseconds(150));
Assert.Equal(1, service.PendingPushCount);
// ...and the next viewer still starts cleanly, picking up the queued snapshot.
int enumerationsBefore = service.EnumerationCount;
using CancellationTokenSource nextCancellation = new();
IAsyncEnumerator<DashboardSnapshot> next =
feed.WatchAsync(nextCancellation.Token).GetAsyncEnumerator(nextCancellation.Token);
Task<bool> nextMove = next.MoveNextAsync().AsTask();
await WaitUntilAsync(() => service.EnumerationCount > enumerationsBefore);
Assert.True(await nextMove.WaitAsync(TestTimeout));
Assert.Equal("orphan-check", next.Current.GatewayVersion);
await nextCancellation.CancelAsync();
await DrainAsync(next, nextMove);
}
/// <summary>Builds a snapshot whose version string identifies it in assertions.</summary>
/// <param name="version">Identity marker carried in <c>GatewayVersion</c>.</param>
/// <returns>A snapshot carrying the supplied identity marker.</returns>
@@ -348,6 +414,12 @@ public sealed class DashboardSnapshotFeedTests
/// <summary>Gets a task that completes when a held enumerator disposal is reached.</summary>
public Task DisposalReached => _disposalReached.Task;
/// <summary>
/// Gets the number of queued snapshots no enumeration has taken yet. A live pump
/// drains this even with nobody watching, so a stable count proves the feed is idle.
/// </summary>
public int PendingPushCount => _pushes.Reader.Count;
/// <summary>Gets the number of enumerations that have finished (cancelled, faulted, or completed).</summary>
public int CompletedEnumerationCount => Volatile.Read(ref _completedEnumerationCount);