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

using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;

/// <summary>
/// WP2.1b — the DataConnectionActor against a batch-capable adapter: chunked subscribe,
/// chunked/bounded re-subscribe, the seed deadline, batched tag-resolution probes, the
/// partial-failure surface and the coalesced quality push. The existing suite covers the
/// same actor against a NON batch-capable adapter, which keeps the per-tag fallback pinned.
/// </summary>
public class DataConnectionActorBatchTests : TestKit
{
    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();

    private static DataConnectionOptions Options() => new()
    {
        ReconnectInterval = TimeSpan.FromMilliseconds(100),
        TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150),
        TagResolutionRetryMaxInterval = TimeSpan.FromMilliseconds(600),
        WriteTimeout = TimeSpan.FromSeconds(5),
        SeedReadMaxAttempts = 1,
        SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
        SubscribeBatchSize = 5,
        SubscribeBatchDelay = TimeSpan.FromMilliseconds(50),
        SeedReadBatchSize = 4,
        SeedReadMaxParallelism = 2,
        SeedOverallTimeout = TimeSpan.FromSeconds(30),
        QualityFlushInterval = TimeSpan.FromMilliseconds(200)
    };

    private IActorRef CreateActor(FakeBatchDataConnection adapter, DataConnectionOptions options, string name) =>
        Sys.ActorOf(Props.Create(() => new DataConnectionActor(
            name, adapter, options, _health, _factory, "OpcUa")), name);

    private static string[] Tags(int count) => Enumerable.Range(1, count).Select(i => $"tag{i}").ToArray();

    [Fact]
    public void Subscribe_IssuesOneAdapterRoundTripPerChunk_NotPerTag()
    {
        // 12 tags at SubscribeBatchSize 5 → 3 batch calls (5/5/2) and ZERO per-tag
        // subscribes. This is finding #3: the old path was one adapter call (and one OPC UA
        // ApplyChanges) per tag.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-chunking");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-chunking", Tags(12), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var batches = adapter.SubscribeBatches.ToList();
        Assert.Equal(3, batches.Count);
        Assert.Equal([5, 5, 2], batches.Select(b => b.Count));
        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
    }

    [Fact]
    public void Seed_UsesChunkedBulkReads_NotPerTagReads()
    {
        // Seeding 6 tags at SeedReadBatchSize 4 → 2 bulk reads, no single-tag reads.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-seed");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed", Tags(6), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var reads = adapter.ReadBatches.ToList();
        Assert.Equal(2, reads.Count);
        Assert.Equal(6, reads.Sum(r => r.Count));
        Assert.Equal(0, Volatile.Read(ref adapter.SingleReadCalls));
        // The seeded value still reaches the instance actor after registration.
        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Good, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void Seed_HangingReads_ReturnAtTheOverallDeadline_AndTheSubscribeIsStillAcked()
    {
        // A device whose reads never answer must not hold SubscribeCompleted (and therefore
        // the instance's ack) open: the whole seed is bounded by SeedOverallTimeout.
        var options = Options();
        options.SeedOverallTimeout = TimeSpan.FromMilliseconds(600);
        options.SeedReadTimeout = TimeSpan.FromSeconds(30); // deliberately longer than the deadline
        options.SeedReadMaxAttempts = 3;

        var adapter = new FakeBatchDataConnection { HangReads = true };
        var actor = CreateActor(adapter, options, "batch-seed-deadline");

        var started = DateTimeOffset.UtcNow;
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed-deadline", Tags(3), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
        var elapsed = DateTimeOffset.UtcNow - started;

        // Bounded by the overall deadline, not by SeedReadTimeout (30s) or the retry budget.
        Assert.True(elapsed < TimeSpan.FromSeconds(5), $"seed took {elapsed.TotalSeconds:F1}s");
        // No value was seeded — the tags stay Uncertain until a change notification.
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    [Fact]
    public void PartialFailure_PerTagRow_MarksTagBad_ButTheRequestStillSucceeds()
    {
        // A per-tag failure row is a tag-resolution problem: Bad quality is pushed for that
        // tag and the subscribe is still acked Success (the surviving tags are live).
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag2");
        var actor = CreateActor(adapter, Options(), "batch-partial");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-partial", Tags(3), DateTimeOffset.UtcNow));

        ExpectMsg<TagValueUpdate>(u => u.TagPath == "tag2" && u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void ThrownBatch_AtConnectionLevel_FailsTheRequestAndDrivesReconnect()
    {
        // A THROWN batch means the whole chunk failed at connection level: the response must
        // say so, and the actor must enter Reconnecting (which pushes bad quality for the
        // connection).
        var adapter = new FakeBatchDataConnection
        {
            BatchSubscribeThrows = () => new InvalidOperationException("client is not connected")
        };
        var actor = CreateActor(adapter, Options(), "batch-connection-fault");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-connection-fault", Tags(3), DateTimeOffset.UtcNow));

        ExpectMsg<SubscribeTagsResponse>(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5));
        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void ResolutionProbes_AreBatched_NotOnePerTag()
    {
        // Every unresolved tag is probed in ONE batch call per chunk. Pre-fix the retry tick
        // fired one SubscribeAsync task per unresolved tag, every tick, forever.
        var options = Options();
        options.SubscribeBatchSize = 10;
        var adapter = new FakeBatchDataConnection();
        foreach (var tag in Tags(6))
            adapter.FailingTags.Add(tag);

        var actor = CreateActor(adapter, options, "batch-probe");
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-probe", Tags(6), DateTimeOffset.UtcNow));

        for (var i = 0; i < 6; i++)
            ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        var initialCalls = adapter.SubscribeBatches.Count;
        AwaitCondition(() => adapter.SubscribeBatches.Count > initialCalls, TimeSpan.FromSeconds(5));

        // The probe round carried all six unresolved tags in a single call.
        var probe = adapter.SubscribeBatches.Skip(initialCalls).First();
        Assert.Equal(6, probe.Count);
        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
    }

    [Fact]
    public void ResolutionProbes_BackOff_SoADeadDeviceIsNotProbedAtFullRate()
    {
        // Floor 150ms, ceiling 600ms: a fixed-interval retry would fire ~13 rounds in 2s;
        // the backoff sequence (150, 300, 600, 600 …) fires far fewer.
        var options = Options();
        options.SubscribeBatchSize = 10;
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag1");

        var actor = CreateActor(adapter, options, "batch-backoff");
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-backoff", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        // This is a genuine real-time rate assertion (how many retry rounds a backoff
        // schedule produces inside a fixed wall-clock window), not a synchronization
        // primitive standing in for "wait until X happens" — proving the floor (>= 2
        // rounds) is unavoidably tied to actually waiting out the window. What the old
        // blind Thread.Sleep(2000) + single end-of-window sample could NOT catch is a
        // regression to full-rate (no-backoff) retrying: it would still land somewhere
        // in [2, 8] by luck if the window elapsed while under load, or it would blow
        // past 8 well before the 2s mark and nobody would know until the final sample.
        // AssertRoundsWithinWindow polls throughout and fails immediately the first time
        // the ceiling is breached, so a no-backoff regression is caught fast instead of
        // being masked by (or racing) the fixed window.
        var initialCalls = adapter.SubscribeBatches.Count;
        AssertRoundsWithinWindow(TimeSpan.FromSeconds(2),
            () => adapter.SubscribeBatches.Count - initialCalls,
            minInclusive: 2, maxInclusive: 8,
            context: "dead-device probe backoff (floor 150ms, ceiling 600ms)");
    }

    /// <summary>
    /// Row-40 house pattern adapted for a RATE assertion: proves a monotonically-increasing
    /// count stays within <c>[minInclusive, maxInclusive]</c> across the whole
    /// <paramref name="window"/>, rather than a single blind sleep-then-sample at the end.
    /// The upper bound is enforced continuously (fails fast the instant it is exceeded,
    /// well before the window elapses); the lower bound can only be confirmed once the
    /// full window has elapsed, since proving "at least N happened by wall-clock time T"
    /// requires actually waiting until T.
    /// </summary>
    private static void AssertRoundsWithinWindow(
        TimeSpan window, Func<int> currentRounds, int minInclusive, int maxInclusive, string context)
    {
        var deadline = DateTime.UtcNow + window;
        while (true)
        {
            var rounds = currentRounds();
            Assert.True(rounds <= maxInclusive,
                $"{context}: {rounds} rounds already fired, exceeding the ceiling of {maxInclusive} " +
                "before the window even elapsed — backoff is not slowing retries down.");
            if (DateTime.UtcNow >= deadline)
            {
                Assert.True(rounds >= minInclusive,
                    $"{context}: only {rounds} rounds fired in the window (need >= {minInclusive}).");
                return;
            }
            Thread.Sleep(20);
        }
    }

    /// <summary>
    /// Row-40 house pattern for an ABSENCE assertion (mirrors the Medium-wave
    /// <c>SiteAlarmAggregatorActorTests.AssertHoldsFor</c> treatment): proves
    /// <paramref name="holds"/> keeps being true for the whole <paramref name="window"/> by
    /// polling, rather than a single blind <c>Thread.Sleep(window)</c> followed by one
    /// end-of-window sample. A bare sleep-then-assert cannot distinguish "the mechanism
    /// correctly stayed quiet" from "the mechanism just hadn't run yet when the one sample
    /// was taken", and it can't catch a value that flips wrong and then flips back before
    /// that single sample. This fails immediately the first time <paramref name="holds"/>
    /// goes false, and only returns once the full window has elapsed with it still true.
    /// </summary>
    private static void AssertHoldsFor(TimeSpan window, Func<bool> holds, string because)
    {
        var deadline = DateTime.UtcNow + window;
        while (true)
        {
            Assert.True(holds(), because);
            if (DateTime.UtcNow >= deadline) return;
            Thread.Sleep(20);
        }
    }

    [Fact]
    public void QualityCounters_ArePushedOnTransitionOnly_AndCoalesced()
    {
        // Repeated values at UNCHANGED quality move no counter and must produce no collector
        // push at all; a genuine transition produces exactly one push per coalescing window.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-quality");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
        AwaitCondition(() => adapter.ValueCallback != null, TimeSpan.FromSeconds(5));

        // Let the seed's own transition flush, then start counting.
        Thread.Sleep(400);
        _health.ClearReceivedCalls();

        for (var i = 0; i < 20; i++)
            adapter.ValueCallback!("tag1", new TagValue(i, QualityCode.Good, DateTimeOffset.UtcNow));

        AssertHoldsFor(TimeSpan.FromMilliseconds(400),
            () => !_health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateTagQuality"),
            "20 unchanged-quality repeats must never push UpdateTagQuality");

        // One genuine transition → exactly one coalesced push.
        adapter.ValueCallback!("tag1", new TagValue(1, QualityCode.Bad, DateTimeOffset.UtcNow));
        AwaitCondition(
            () => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagQuality") == 1,
            TimeSpan.FromSeconds(3));
        AssertHoldsFor(TimeSpan.FromMilliseconds(300),
            () => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagQuality") == 1,
            "the coalesced push must not repeat");
    }

    [Fact]
    public void QualityCounters_FlushImmediatelyOnDisconnect()
    {
        // "Immediate bad quality on disconnect" must never wait for the coalescing timer.
        var options = Options();
        options.QualityFlushInterval = TimeSpan.FromSeconds(30); // would mask a deferred push
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, options, "batch-quality-disconnect");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality-disconnect", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
        ExpectMsg<TagValueUpdate>(TimeSpan.FromSeconds(5)); // the seed
        _health.ClearReceivedCalls();

        adapter.RaiseDisconnected();

        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        AwaitCondition(
            () => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateTagQuality"),
            TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Unsubscribe_ReleasesEveryHandleInOneRoundTrip()
    {
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-unsubscribe");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-unsubscribe", Tags(6), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-unsubscribe", DateTimeOffset.UtcNow));

        AwaitCondition(() => adapter.UnsubscribeBatches.Count == 1, TimeSpan.FromSeconds(5));
        Assert.True(adapter.UnsubscribeBatches.TryDequeue(out var released));
        Assert.Equal(6, released!.Count);
    }

    [Fact]
    public void Reconnect_ReSubscribesInPacedChunks_AndRestoresTheResolvedCount()
    {
        // The reconnect re-subscribe is chunked and paced (SubscribeBatchDelay between
        // chunks) rather than firing one task per tag, and the per-tag results repopulate
        // _subscriptionIds so a later unsubscribe still releases every adapter handle.
        var options = Options();
        options.SubscribeBatchSize = 4;
        options.SubscribeBatchDelay = TimeSpan.FromMilliseconds(120);

        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, options, "batch-resubscribe");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-resubscribe", Tags(8), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var beforeReconnect = adapter.SubscribeBatches.Count;
        adapter.RaiseDisconnected();

        // Two re-subscribe chunks of 4.
        AwaitCondition(() => adapter.SubscribeBatches.Count >= beforeReconnect + 2, TimeSpan.FromSeconds(10));
        var reconnectChunks = adapter.SubscribeBatches.Skip(beforeReconnect).Take(2).ToList();
        Assert.Equal([4, 4], reconnectChunks.Select(c => c.Count));

        // Paced: the second chunk is not issued back-to-back with the first.
        var times = adapter.SubscribeBatchTimes.Skip(beforeReconnect).Take(2).ToList();
        Assert.True(times[1] - times[0] >= TimeSpan.FromMilliseconds(100),
            $"chunks were {(times[1] - times[0]).TotalMilliseconds:F0}ms apart");

        // Every tag is registered again, so unsubscribe releases all 8 handles.
        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-resubscribe", DateTimeOffset.UtcNow));
        AwaitCondition(
            () => 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));
    }

    // ── 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)
    {
        actor.Tell(new DataConnectionActor.GetHealthReport());
        return FishForMessage<DataConnectionHealthReport>(_ => true, TimeSpan.FromSeconds(5));
    }
}