fix(dcl): discard in-flight subscribe results for unsubscribed tags; release the orphaned handle

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<string>();
+
         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<string>();
+        var idsToRelease = new List<string>();
 
         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

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<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
+        ExpectMsg<SubscribeTagsResponse>(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<SubscribeTagsResponse>(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<SubscribeTagsResponse>(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));
+    }
+
+    /// <summary>Point-in-time counters straight off the actor, fished past any pending pushes.</summary>
+    private DataConnectionHealthReport Health(IActorRef actor)
+    {
+        actor.Tell(new DataConnectionActor.GetHealthReport());
+        return FishForMessage<DataConnectionHealthReport>(_ => true, TimeSpan.FromSeconds(5));
+    }
 }

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs

@@ -35,6 +35,13 @@ public sealed class FakeBatchDataConnection
     public Func<Exception>? BatchSubscribeThrows;
     /// <summary>When true, bulk reads never return until the caller's token cancels.</summary>
     public bool HangReads;
+    /// <summary>
+    /// When set, <see cref="SubscribeBatchAsync"/> 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).
+    /// </summary>
+    public Task? SubscribeGate;
     /// <summary>Value returned for every readable tag.</summary>
     public object? SeedValue = 42;
 
@@ -67,7 +74,7 @@ public sealed class FakeBatchDataConnection
     }
 
     /// <inheritdoc />
-    public Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
+    public async Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
         IReadOnlyList<string> 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<TagSubscribeResult> 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;
     }
 
     /// <inheritdoc />