From 37f13e2eaa290c3b400a37d7b5d7ca9883a40ab0 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 14 Aug 2026 23:25:48 -0400 Subject: [PATCH] fix(dcl): discard in-flight subscribe results for unsubscribed tags; release the orphaned handle --- .../Actors/DataConnectionActor.cs | 66 ++++++++- .../Actors/DataConnectionActorBatchTests.cs | 135 ++++++++++++++++++ .../Actors/FakeBatchDataConnection.cs | 14 +- 3 files changed, 208 insertions(+), 7 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index a5f15429..9dfff812 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -1152,6 +1152,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // request be re-stashed/retried after reconnect via ReSubscribeAll. var connectionLevelFailure = msg.Results.Any(r => !r.Success && r.ConnectionLevelFailure); + // Handles this request created that turn out to be redundant; released in ONE + // round trip below, symmetric with HandleBatchSubscribeCompleted. + var redundantIds = new List(); + foreach (var result in msg.Results) { // A result with AlreadySubscribed: false means @@ -1175,9 +1179,26 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // Re-check against current state: another subscribe may have resolved the // same tag while this request's I/O was in flight. - if (result.AlreadySubscribed || _subscriptionIds.ContainsKey(result.TagPath)) + // AlreadySubscribed rows are partitioned out in HandleSubscribe and carry NO + // SubscriptionId of their own (another caller owns the adapter handle), so + // there is nothing to release for them. + if (result.AlreadySubscribed) continue; + if (_subscriptionIds.ContainsKey(result.TagPath)) + { + // Another path (a resolution probe, a reconnect re-subscribe or a + // concurrent request for a different instance) stored a handle for this + // tag while this request's I/O was in flight. This request issued its OWN + // SubscribeAsync, so dropping the row without releasing its handle leaks a + // monitored item forever — _subscriptionIds holds a single id per tag, so + // no later unsubscribe can ever reference this one. Release it, exactly as + // HandleBatchSubscribeCompleted does for its duplicate rows. + if (result is { Success: true, SubscriptionId: not null }) + redundantIds.Add(result.SubscriptionId); + continue; + } + if (result.Success) { _subscriptionIds[result.TagPath] = result.SubscriptionId!; @@ -1239,6 +1260,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers } } + // Fire-and-forget release of every handle this request created redundantly. + _ = UnsubscribeIdsAsync(_adapter, redundantIds); + // Now that every tag is registered in // _subscriptionsByInstance, deliver the values captured by the initial read. // Re-entering via Self reuses HandleTagValueReceived's generation guard, fan-out @@ -1897,7 +1921,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers } var anyResolved = false; - var duplicateIds = new List(); + var idsToRelease = new List(); foreach (var row in msg.Results) { @@ -1906,12 +1930,37 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers if (row is { Success: true, SubscriptionId: not null }) { var wasUnresolved = _unresolvedTags.Remove(row.TagPath); + + // The tag lost its last subscriber while this batch was in flight — an + // instance unsubscribe (disable/undeploy/redeploy) raced the resolution + // probe or the reconnect re-subscribe. _instancesByTag is the authority: + // it is the inverse of _subscriptionsByInstance and, like it, is preserved + // across reconnect, so an empty entry means nobody wants this tag any more. + // Applying the row anyway would store a handle no future unsubscribe can + // ever release (HandleUnsubscribe already ran for this tag) AND push + // _resolvedTags above the _totalSubscribed that same unsubscribe just + // decremented — permanently corrupting the health counters and driving + // _totalSubscribed negative on the next redeploy round trip. So discard + // every state mutation for this row and release the adapter handle the + // batch just created. This restores the gate the pre-merge per-tag handler + // had (it applied only when the tag was still in _unresolvedTags) and + // additionally releases the handle that path leaked. + if (!_instancesByTag.ContainsKey(row.TagPath)) + { + _log.Debug( + "[{0}] Discarding batch-subscribe result for {1} — the tag was " + + "unsubscribed while the subscribe was in flight; releasing its handle.", + _connectionName, row.TagPath); + idsToRelease.Add(row.SubscriptionId); + continue; + } + if (_subscriptionIds.ContainsKey(row.TagPath)) { // Another path already stored a handle for this tag while this one was // in flight — release the redundant handle instead of leaking it // (mirrors the duplicate-alarm-feed guard). - duplicateIds.Add(row.SubscriptionId); + idsToRelease.Add(row.SubscriptionId); } else { @@ -1933,11 +1982,18 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers { _log.Debug("[{0}] Tag resolution still failing for {1}: {2}", _connectionName, row.TagPath, row.Error); - _unresolvedTags.Add(row.TagPath); + + // Same in-flight-unsubscribe race as the success branch above: the tag was + // already dropped from _unresolvedTags AND from _totalSubscribed by + // HandleUnsubscribe, so re-adding it here would probe a tag nobody + // subscribes to forever, at a retry count TotalSubscribedTags no longer + // accounts for. + if (_instancesByTag.ContainsKey(row.TagPath)) + _unresolvedTags.Add(row.TagPath); } } - _ = UnsubscribeIdsAsync(_adapter, duplicateIds); + _ = UnsubscribeIdsAsync(_adapter, idsToRelease); _healthCollector.UpdateTagResolution(_connectionName, _totalSubscribed, _resolvedTags); // Backoff bookkeeping belongs to the probe round only: a reconnect re-subscribe diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionActorBatchTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionActorBatchTests.cs index e681de59..b5211b8c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionActorBatchTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionActorBatchTests.cs @@ -288,4 +288,139 @@ public class DataConnectionActorBatchTests : TestKit () => adapter.UnsubscribeBatches.Any(b => b.Count == 8), TimeSpan.FromSeconds(10)); } + + [Fact] + public void ResolutionProbe_CompletingAfterUnsubscribe_IsDiscarded_AndItsHandleIsReleased() + { + // A tag unsubscribed while its resolution probe is in flight must NOT have the + // probe's result applied: HandleUnsubscribe already dropped it from + // _unresolvedTags and from TotalSubscribedTags, so storing the handle would push + // ResolvedTags above TotalSubscribedTags forever (and drive TotalSubscribedTags + // negative on the next redeploy) while leaking the adapter monitored item, which + // no later unsubscribe could ever reference. + var options = Options(); + options.SubscribeBatchSize = 10; + // Long enough that the probe cannot fire before the gate below is armed. + options.TagResolutionRetryInterval = TimeSpan.FromMilliseconds(800); + options.TagResolutionRetryMaxInterval = TimeSpan.FromSeconds(2); + + var adapter = new FakeBatchDataConnection(); + adapter.FailingTags.Add("tag1"); + var actor = CreateActor(adapter, options, "batch-probe-unsubscribe-race"); + + actor.Tell(new SubscribeTagsRequest( + "c1", "inst1", "batch-probe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow)); + ExpectMsg(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5)); + ExpectMsg(TimeSpan.FromSeconds(5)); + + var afterSubscribe = Health(actor); + Assert.Equal(1, afterSubscribe.TotalSubscribedTags); + Assert.Equal(0, afterSubscribe.ResolvedTags); + + // Park the next probe inside the adapter, and let it succeed when released. + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + adapter.SubscribeGate = gate.Task; + adapter.FailingTags.Clear(); + + var beforeProbe = adapter.SubscribeBatches.Count; + AwaitCondition(() => adapter.SubscribeBatches.Count > beforeProbe, TimeSpan.FromSeconds(10)); + + // Unsubscribe while the probe is still in flight, and confirm it has been applied + // before the probe result lands (same-sender ordering makes the report a barrier). + actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-probe-unsubscribe-race", DateTimeOffset.UtcNow)); + var afterUnsubscribe = Health(actor); + Assert.Equal(0, afterUnsubscribe.TotalSubscribedTags); + Assert.Equal(0, afterUnsubscribe.ResolvedTags); + // The tag was still unresolved, so the unsubscribe itself had no handle to + // release — every id released from here on is the raced probe's. (This also + // rules out a vacuous run in which the probe resolved before the gate was armed.) + Assert.Empty(adapter.UnsubscribeBatches); + + gate.SetResult(); + + // The orphaned handle is released back to the adapter… + AwaitCondition( + () => adapter.UnsubscribeBatches.Any(b => b.Count == 1), + TimeSpan.FromSeconds(10)); + + // …and the counters are untouched by the discarded result. + AwaitAssert(() => + { + var report = Health(actor); + Assert.Equal(0, report.TotalSubscribedTags); + Assert.Equal(0, report.ResolvedTags); + }, TimeSpan.FromSeconds(5)); + + // A redeploy round trip still books exactly one tag in and one tag out — pre-fix + // the stale handle made this read Total=0/Resolved=1 and then Total=-1. + adapter.SubscribeGate = null; + actor.Tell(new SubscribeTagsRequest( + "c3", "inst1", "batch-probe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow)); + ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); + + var afterRedeploy = Health(actor); + Assert.Equal(1, afterRedeploy.TotalSubscribedTags); + Assert.Equal(1, afterRedeploy.ResolvedTags); + + actor.Tell(new UnsubscribeTagsRequest("c4", "inst1", "batch-probe-unsubscribe-race", DateTimeOffset.UtcNow)); + var afterFinalUnsubscribe = Health(actor); + Assert.Equal(0, afterFinalUnsubscribe.TotalSubscribedTags); + Assert.Equal(0, afterFinalUnsubscribe.ResolvedTags); + } + + [Fact] + public void ReconnectResubscribe_CompletingAfterUnsubscribe_IsDiscarded_AndItsHandleIsReleased() + { + // Same race on the OTHER batch source: the reconnect re-subscribe. ReSubscribeAll + // has already zeroed ResolvedTags, so applying the late result would report a + // resolved tag for an instance that no longer exists and leak its handle. + var options = Options(); + options.SubscribeBatchSize = 10; + + var adapter = new FakeBatchDataConnection(); + var actor = CreateActor(adapter, options, "batch-resubscribe-unsubscribe-race"); + + actor.Tell(new SubscribeTagsRequest( + "c1", "inst1", "batch-resubscribe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow)); + ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); + + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + adapter.SubscribeGate = gate.Task; + + var beforeReconnect = adapter.SubscribeBatches.Count; + adapter.RaiseDisconnected(); + AwaitCondition(() => adapter.SubscribeBatches.Count > beforeReconnect, TimeSpan.FromSeconds(10)); + + actor.Tell(new UnsubscribeTagsRequest( + "c2", "inst1", "batch-resubscribe-unsubscribe-race", DateTimeOffset.UtcNow)); + // The unsubscribe found no handle to release (ReSubscribeAll cleared + // _subscriptionIds), so no release round trip has happened yet. + Health(actor); + Assert.Empty(adapter.UnsubscribeBatches); + + gate.SetResult(); + + // The handle minted by the re-subscribe is released rather than leaked, and the + // discarded row books no resolved tag. (TotalSubscribedTags across a reconnect + // window is a separate, pre-existing accounting gap — ReSubscribeAll clears the + // maps HandleUnsubscribe decrements from — so this pins the invariant that + // matters here: ResolvedTags never exceeds it.) + AwaitCondition( + () => adapter.UnsubscribeBatches.Any(b => b.Count == 1), + TimeSpan.FromSeconds(10)); + + AwaitAssert(() => + { + var report = Health(actor); + Assert.Equal(0, report.ResolvedTags); + Assert.True(report.ResolvedTags <= report.TotalSubscribedTags); + }, TimeSpan.FromSeconds(5)); + } + + /// Point-in-time counters straight off the actor, fished past any pending pushes. + private DataConnectionHealthReport Health(IActorRef actor) + { + actor.Tell(new DataConnectionActor.GetHealthReport()); + return FishForMessage(_ => true, TimeSpan.FromSeconds(5)); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs index ad4067c9..6c93c92b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs @@ -35,6 +35,13 @@ public sealed class FakeBatchDataConnection public Func? BatchSubscribeThrows; /// When true, bulk reads never return until the caller's token cancels. public bool HangReads; + /// + /// When set, records the call and then parks until this + /// task completes, before producing its result rows. Lets a test hold a subscribe batch + /// in flight while it drives other messages into the actor (e.g. an unsubscribe that + /// races the completion). + /// + public Task? SubscribeGate; /// Value returned for every readable tag. public object? SeedValue = 42; @@ -67,7 +74,7 @@ public sealed class FakeBatchDataConnection } /// - public Task> SubscribeBatchAsync( + public async Task> SubscribeBatchAsync( IReadOnlyList tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default) { SubscribeBatches.Enqueue(tagPaths.ToList()); @@ -77,12 +84,15 @@ public sealed class FakeBatchDataConnection if (BatchSubscribeThrows is { } factory) throw factory(); + if (SubscribeGate is { } gate) + await gate; + IReadOnlyList rows = tagPaths .Select(t => FailingTags.Contains(t) ? new TagSubscribeResult(t, false, null, "node not found") : new TagSubscribeResult(t, true, $"sub-{Interlocked.Increment(ref _nextId)}", null)) .ToList(); - return Task.FromResult(rows); + return rows; } ///