From 756296886b6baf1f9651057d44808e6e4e31c1d8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 20:43:17 -0400 Subject: [PATCH] =?UTF-8?q?fix(dashboard):=20generation-scoped=20idle=20ga?= =?UTF-8?q?te=20in=20UnsubscribeAsync=20=E2=80=94=20no=20zero-subscriber?= =?UTF-8?q?=20pump=20survives?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Dashboard/DashboardSnapshotFeed.cs | 42 +++++++++-- .../Dashboard/DashboardSnapshotFeedTests.cs | 72 +++++++++++++++++++ 2 files changed, 109 insertions(+), 5 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs index ede8f0e..301104f 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardSnapshotFeed.cs @@ -14,8 +14,9 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard; /// /// /// 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 +/// awaited when the last subscriber of the live generation 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. /// @@ -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 } } + /// Reports whether any subscriber is still being served by a generation. + /// The generation to look for. + /// True when at least one subscriber carries that generation. + private bool HasSubscribersLocked(long generation) + { + foreach (Subscription subscriber in _subscribers) + { + if (subscriber.Generation == generation) + { + return true; + } + } + + return false; + } + /// /// Starts a pump generation. Must be called while holding _gate; the caller adds /// the subscribers that belong to the returned 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 330ac17..8c0091e 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotFeedTests.cs @@ -254,6 +254,72 @@ public sealed class DashboardSnapshotFeedTests await DrainAsync(second, secondMove); } + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WatchAsync_WhenAJoinerLeavesWhileAnOldGenerationLingers_LeavesNoPumpRunning() + { + FakeSnapshotService service = new(); + service.HoldDisposal(); + DashboardSnapshotFeed feed = new(service); + + using CancellationTokenSource oldCancellation = new(); + IAsyncEnumerator oldSubscriber = + feed.WatchAsync(oldCancellation.Token).GetAsyncEnumerator(oldCancellation.Token); + Task 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 joiner = + feed.WatchAsync(joinerCancellation.Token).GetAsyncEnumerator(joinerCancellation.Token); + Task 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(() => 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 next = + feed.WatchAsync(nextCancellation.Token).GetAsyncEnumerator(nextCancellation.Token); + Task 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); + } + /// Builds a snapshot whose version string identifies it in assertions. /// Identity marker carried in GatewayVersion. /// A snapshot carrying the supplied identity marker. @@ -348,6 +414,12 @@ public sealed class DashboardSnapshotFeedTests /// Gets a task that completes when a held enumerator disposal is reached. public Task DisposalReached => _disposalReached.Task; + /// + /// 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. + /// + public int PendingPushCount => _pushes.Reader.Count; + /// Gets the number of enumerations that have finished (cancelled, faulted, or completed). public int CompletedEnumerationCount => Volatile.Read(ref _completedEnumerationCount);