fix(dcl): derive tag-resolution health counts from per-tag authoritative state

Closes arch-review remediation residual #1 (DCL unsubscribe-during-reconnect count staleness). DataConnectionActor tracked TotalSubscribedTags/ResolvedTags as two int fields incremented and decremented at five independent sites. ReSubscribeAll clears the very maps those decrements key off (_subscriptionIds, _unresolvedTags) while deliberately preserving _subscriptionsByInstance, so an unsubscribe landing inside a reconnect window matched NEITHER decrement branch: the total leaked +1 per subscribe/reconnect/unsubscribe churn cycle, permanently and cumulatively. The 37f13e2e discard gate stopped the orphan-handle half of that race; it could not stop the counters drifting, because they were state of their own. Both counts are now DERIVED at report time from the authoritative per-tag collections, which makes the drift unrepresentable rather than merely guarded: total = _instancesByTag.Count (the per-tag counted set the residual called for — distinct tags with at least one subscribing instance) resolved = _subscriptionIds.Count (tags for which the adapter holds a handle) Two semantic corrections fall out of the derivation: - A tag whose subscribe failed at CONNECTION level now counts toward the total. It was excluded before, yet the reconnect re-subscribe re-issued it from _subscriptionsByInstance and booked it as resolved — resolved above total, and a total driven negative by the eventual unsubscribe. - _tagSubscriberCount is deleted. It duplicated _instancesByTag exactly, so HandleUnsubscribe's last-subscriber test is now "did UnindexTag drop the key?" — still O(1), with no parallel count that can disagree about when a handle is released. The subscribe-success promotion split (fresh vs. unresolved→resolved) also goes: it existed only to pick which scalar to bump; set sizes get DataConnectionLayer-020's double-count cases right for free. Behavior is otherwise unchanged — same logging, same handle release, same unresolved-tag probing, same in-flight-unsubscribe discard semantics (the long comment block there is updated for the mechanics that changed). Tests: five TagResolutionCounts_* cases in DataConnectionActorBatchTests covering the churn repro (3 cycles), a shared tag losing one instance mid reconnect, connection-level failure then recovery, plain subscribe/unsubscribe cycles, and a completed reconnect re-subscribe. Verified failing against the pre-fix actor (churn: total 1 not 0; connection-level: total 0 not 1) and passing after. Full DCL suite 319/319; solution builds with 0 warnings. Docs: Component-DataConnectionLayer.md health-reporting section describes the derived counts; residuals register item 1 marked RESOLVED.
2026-08-15 02:05:35 -04:00
parent 986e6e7ad5
commit 491df111ea
5 changed files with 389 additions and 125 deletions
@@ -417,6 +417,244 @@ public class DataConnectionActorBatchTests : TestKit
        }, TimeSpan.FromSeconds(5));
    }

+    // ── Tag-resolution health counts: derived from the per-tag counted set ──
+    //
+    // Arch-review remediation residual #1. TotalSubscribedTags / ResolvedTags used to be
+    // two int fields incremented and decremented at five independent sites. The reconnect
+    // path cleared the very maps those decrements keyed off (_subscriptionIds and
+    // _unresolvedTags), so an unsubscribe landing inside a reconnect window matched NEITHER
+    // decrement branch and leaked a phantom tag into the total — permanently, and cumulative
+    // under churn. They are now derived at report time from _instancesByTag (the per-tag
+    // counted set) and _subscriptionIds, which makes that class of drift unrepresentable.
+
+    [Fact]
+    public void TagResolutionCounts_UnsubscribeInsideAReconnectWindow_DoesNotLeakASubscribedTag()
+    {
+        // THE residual repro. Each cycle: subscribe → drop the connection → hold the
+        // reconnect re-subscribe in flight → unsubscribe inside that window → let the batch
+        // land. Pre-fix the total read 1 after the first cycle's unsubscribe (nothing
+        // decremented it) and then 2, 3 … at the head of each following cycle — a +1 leak
+        // per churn round that never recovers. Post-fix every cycle books exactly one tag
+        // in and one tag out.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, options, "counts-reconnect-churn");
+
+        for (var cycle = 1; cycle <= 3; cycle++)
+        {
+            actor.Tell(new SubscribeTagsRequest(
+                $"s{cycle}", "inst1", "counts-reconnect-churn", ["tag1"], DateTimeOffset.UtcNow));
+            FishForMessage<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+
+            var afterSubscribe = Health(actor);
+            Assert.Equal(1, afterSubscribe.TotalSubscribedTags);
+            Assert.Equal(1, afterSubscribe.ResolvedTags);
+
+            // Park the reconnect re-subscribe: ReSubscribeAll has already cleared
+            // _subscriptionIds and _unresolvedTags, and the batch has not come back yet.
+            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));
+
+            // Unsubscribe INSIDE the reconnect window (same-sender ordering makes the
+            // health report a barrier proving it was applied first).
+            actor.Tell(new UnsubscribeTagsRequest(
+                $"u{cycle}", "inst1", "counts-reconnect-churn", DateTimeOffset.UtcNow));
+
+            var afterUnsubscribe = Health(actor);
+            Assert.Equal(0, afterUnsubscribe.TotalSubscribedTags);
+            Assert.Equal(0, afterUnsubscribe.ResolvedTags);
+
+            // Release the raced batch. Its rows are discarded and its handle released —
+            // one release per cycle, which is also the barrier that the discard ran.
+            gate.SetResult();
+            adapter.SubscribeGate = null;
+            AwaitCondition(() => adapter.UnsubscribeBatches.Count >= cycle, TimeSpan.FromSeconds(10));
+
+            var afterDiscard = Health(actor);
+            Assert.Equal(0, afterDiscard.TotalSubscribedTags);
+            Assert.Equal(0, afterDiscard.ResolvedTags);
+        }
+    }
+
+    [Fact]
+    public void TagResolutionCounts_SharedTagUnsubscribedInsideAReconnectWindow_StillCountsForTheOtherInstance()
+    {
+        // The counted-set semantics, not a bare reference count: one instance leaving during
+        // a reconnect window must leave the tag counted (the other instance still subscribes)
+        // and must not release its handle.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+
+        var adapter = new FakeBatchDataConnection();
+        var actor = CreateActor(adapter, options, "counts-shared-reconnect");
+
+        actor.Tell(new SubscribeTagsRequest("s1", "instA", "counts-shared-reconnect", ["tag1"], DateTimeOffset.UtcNow));
+        FishForMessage<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+        actor.Tell(new SubscribeTagsRequest("s2", "instB", "counts-shared-reconnect", ["tag1"], DateTimeOffset.UtcNow));
+        FishForMessage<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+
+        var shared = Health(actor);
+        Assert.Equal(1, shared.TotalSubscribedTags);
+        Assert.Equal(1, shared.ResolvedTags);
+
+        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("u1", "instA", "counts-shared-reconnect", DateTimeOffset.UtcNow));
+        var afterOneLeft = Health(actor);
+        Assert.Equal(1, afterOneLeft.TotalSubscribedTags);
+        Assert.Equal(0, afterOneLeft.ResolvedTags); // ReSubscribeAll cleared the handles
+
+        // instB still wants the tag, so the raced batch row is APPLIED (not discarded) and
+        // nothing is released back to the adapter.
+        gate.SetResult();
+        adapter.SubscribeGate = null;
+        AwaitAssert(() =>
+        {
+            var report = Health(actor);
+            Assert.Equal(1, report.TotalSubscribedTags);
+            Assert.Equal(1, report.ResolvedTags);
+        }, TimeSpan.FromSeconds(10));
+        Assert.Empty(adapter.UnsubscribeBatches);
+
+        // The last subscriber leaving drops the tag from both counts and releases it.
+        actor.Tell(new UnsubscribeTagsRequest("u2", "instB", "counts-shared-reconnect", DateTimeOffset.UtcNow));
+        var afterBothLeft = Health(actor);
+        Assert.Equal(0, afterBothLeft.TotalSubscribedTags);
+        Assert.Equal(0, afterBothLeft.ResolvedTags);
+        AwaitCondition(() => adapter.UnsubscribeBatches.Any(b => b.Count == 1), TimeSpan.FromSeconds(10));
+    }
+
+    [Fact]
+    public void TagResolutionCounts_ConnectionLevelFailureThenRecovery_KeepResolvedWithinTheTotal()
+    {
+        // A tag whose FIRST subscribe fails at connection level is subscribed by an instance
+        // (it is re-issued from _subscriptionsByInstance on reconnect) but is neither
+        // resolved nor unresolved. The retired scalar counted it in neither, so once the
+        // reconnect re-subscribe resolved it the report read ResolvedTags=1 against
+        // TotalSubscribedTags=0 — resolved above total, and a total driven negative by the
+        // eventual unsubscribe.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+
+        var calls = 0;
+        var adapter = new FakeBatchDataConnection
+        {
+            // Fail the initial subscribe at connection level; let the reconnect through.
+            BatchSubscribeThrows = () => Interlocked.Increment(ref calls) == 1
+                ? new InvalidOperationException("client is not connected")
+                : null
+        };
+        var actor = CreateActor(adapter, options, "counts-connection-level");
+
+        actor.Tell(new SubscribeTagsRequest(
+            "s1", "inst1", "counts-connection-level", ["tag1"], DateTimeOffset.UtcNow));
+        FishForMessage<SubscribeTagsResponse>(m => !m.Success, TimeSpan.FromSeconds(10));
+
+        // The tag is subscribed by the instance from the moment it is registered, whatever
+        // the adapter said — so it counts toward the total and not toward resolved.
+        var afterFault = Health(actor);
+        Assert.Equal(1, afterFault.TotalSubscribedTags);
+        Assert.Equal(0, afterFault.ResolvedTags);
+
+        // The reconnect re-subscribe resolves it: 1/1, never 1/0-then-resolved-above-total.
+        AwaitAssert(() =>
+        {
+            var report = Health(actor);
+            Assert.Equal(1, report.TotalSubscribedTags);
+            Assert.Equal(1, report.ResolvedTags);
+        }, TimeSpan.FromSeconds(15));
+
+        actor.Tell(new UnsubscribeTagsRequest("u1", "inst1", "counts-connection-level", DateTimeOffset.UtcNow));
+        var afterUnsubscribe = Health(actor);
+        Assert.Equal(0, afterUnsubscribe.TotalSubscribedTags);
+        Assert.Equal(0, afterUnsubscribe.ResolvedTags);
+    }
+
+    [Fact]
+    public void TagResolutionCounts_SurvivePlainSubscribeUnsubscribeCycles()
+    {
+        // The undramatic path, pinned so the derived counts cannot regress on it: repeated
+        // deploy/undeploy round trips with a mix of resolved and unresolved tags always read
+        // back the same numbers, with no accumulation across cycles.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+        // Long enough that a resolution probe cannot resolve tag4 mid-assertion.
+        options.TagResolutionRetryInterval = TimeSpan.FromSeconds(30);
+        options.TagResolutionRetryMaxInterval = TimeSpan.FromSeconds(60);
+
+        var adapter = new FakeBatchDataConnection();
+        adapter.FailingTags.Add("tag4");
+        var actor = CreateActor(adapter, options, "counts-plain-cycles");
+
+        for (var cycle = 1; cycle <= 3; cycle++)
+        {
+            actor.Tell(new SubscribeTagsRequest(
+                $"s{cycle}", "inst1", "counts-plain-cycles", Tags(4), DateTimeOffset.UtcNow));
+            FishForMessage<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+
+            var afterSubscribe = Health(actor);
+            Assert.Equal(4, afterSubscribe.TotalSubscribedTags); // 3 resolved + 1 unresolved
+            Assert.Equal(3, afterSubscribe.ResolvedTags);
+
+            actor.Tell(new UnsubscribeTagsRequest(
+                $"u{cycle}", "inst1", "counts-plain-cycles", DateTimeOffset.UtcNow));
+
+            var afterUnsubscribe = Health(actor);
+            Assert.Equal(0, afterUnsubscribe.TotalSubscribedTags);
+            Assert.Equal(0, afterUnsubscribe.ResolvedTags);
+        }
+    }
+
+    [Fact]
+    public void TagResolutionCounts_AreCarriedThroughACompletedReconnectReSubscribe()
+    {
+        // A completed reconnect re-subscribe restores the resolved count and leaves the
+        // total exactly where it was — the tags never stopped being subscribed.
+        var options = Options();
+        options.SubscribeBatchSize = 10;
+
+        var adapter = new FakeBatchDataConnection();
+        adapter.FailingTags.Add("tag5");
+        var actor = CreateActor(adapter, options, "counts-reconnect-restore");
+
+        actor.Tell(new SubscribeTagsRequest(
+            "s1", "inst1", "counts-reconnect-restore", Tags(5), DateTimeOffset.UtcNow));
+        FishForMessage<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
+
+        var beforeReconnect = Health(actor);
+        Assert.Equal(5, beforeReconnect.TotalSubscribedTags);
+        Assert.Equal(4, beforeReconnect.ResolvedTags);
+
+        adapter.RaiseDisconnected();
+
+        AwaitAssert(() =>
+        {
+            var report = Health(actor);
+            Assert.Equal(5, report.TotalSubscribedTags);
+            Assert.Equal(4, report.ResolvedTags);
+        }, TimeSpan.FromSeconds(15));
+
+        // The still-unresolved tag resolving on a later probe moves ONLY the resolved count.
+        adapter.FailingTags.Clear();
+        AwaitAssert(() =>
+        {
+            var report = Health(actor);
+            Assert.Equal(5, report.TotalSubscribedTags);
+            Assert.Equal(5, report.ResolvedTags);
+        }, TimeSpan.FromSeconds(15));
+    }
+
    /// <summary>Point-in-time counters straight off the actor, fished past any pending pushes.</summary>
    private DataConnectionHealthReport Health(IActorRef actor)
    {
@@ -31,8 +31,13 @@ public sealed class FakeBatchDataConnection

    /// <summary>Tags reported as failed rows (per-tag resolution failure).</summary>
    public readonly HashSet<string> FailingTags = new(StringComparer.Ordinal);
-    /// <summary>When set, every batch subscribe throws this — a batch-level fault.</summary>
-    public Func<Exception>? BatchSubscribeThrows;
+    /// <summary>
+    /// When set, a batch subscribe throws whatever this returns — a batch-level fault.
+    /// Returning <c>null</c> lets that call through, so a test can make the fault TRANSIENT
+    /// (e.g. fail the initial subscribe at connection level, then let the reconnect
+    /// re-subscribe succeed).
+    /// </summary>
+    public Func<Exception?>? BatchSubscribeThrows;
    /// <summary>When true, bulk reads never return until the caller's token cancels.</summary>
    public bool HangReads;
    /// <summary>
@@ -81,8 +86,8 @@ public sealed class FakeBatchDataConnection
        SubscribeBatchTimes.Enqueue(DateTimeOffset.UtcNow);
        ValueCallback = callback;

-        if (BatchSubscribeThrows is { } factory)
-            throw factory();
+        if (BatchSubscribeThrows?.Invoke() is { } fault)
+            throw fault;

        if (SubscribeGate is { } gate)
            await gate;