ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploySiteAsyncTests.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;

namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests;

/// <summary>
/// WP2.5: bulk site deployment. Pins the properties that make the fan-out safe —
/// concurrency is BOUNDED by the configured degree, one slow instance does not stop
/// the others completing, a per-instance timeout contains a wedged instance, and
/// every instance still gets its own deployment id and its own operation lock.
/// </summary>
public class DeploySiteAsyncTests : TestKit
{
    private const int SiteId = 1;

    private readonly IDeploymentManagerRepository _repo = Substitute.For<IDeploymentManagerRepository>();
    private readonly IFlatteningPipeline _pipeline = Substitute.For<IFlatteningPipeline>();
    private readonly ISiteRepository _siteRepo = Substitute.For<ISiteRepository>();
    private readonly IAuditService _audit = Substitute.For<IAuditService>();
    private readonly IStaleInstanceProbe _staleProbe = Substitute.For<IStaleInstanceProbe>();
    private readonly OperationLockManager _lockManager = new();

    public DeploySiteAsyncTests()
    {
        _siteRepo.GetSiteByIdAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
            .Returns(ci => new Site($"Site {ci.ArgAt<int>(0)}", $"site-{ci.ArgAt<int>(0)}") { Id = ci.ArgAt<int>(0) });

        _pipeline.CreateSession().Returns(_ => new FlattenSession(new TemplateGraphWatermark()));
    }

    /// <summary>Seeds <paramref name="count"/> deployable instances at the test site.</summary>
    /// <param name="count">How many instances to seed.</param>
    /// <param name="disabledIds">Ids seeded <c>Disabled</c> instead of <c>NotDeployed</c>.</param>
    private List<Instance> ArrangeInstances(int count, params int[] disabledIds)
    {
        var instances = new List<Instance>();
        for (var i = 1; i <= count; i++)
        {
            var instance = new Instance($"Inst-{i:00}")
            {
                Id = i,
                SiteId = SiteId,
                TemplateId = 10,
                State = disabledIds.Contains(i) ? InstanceState.Disabled : InstanceState.NotDeployed
            };
            instances.Add(instance);

            _repo.GetInstanceByIdAsync(i, Arg.Any<CancellationToken>()).Returns(instance);
            _repo.GetCurrentDeploymentStatusAsync(i, Arg.Any<CancellationToken>()).Returns((DeploymentRecord?)null);

            var config = new FlattenedConfiguration { InstanceUniqueName = instance.UniqueName };
            _pipeline.FlattenAndValidateAsync(
                    i, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
                .Returns(Result<FlatteningPipelineResult>.Success(
                    new FlatteningPipelineResult(config, $"sha256:{i}", ValidationResult.Success())));
        }

        _siteRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any<CancellationToken>()).Returns(instances);
        return instances;
    }

    private DeploymentService CreateService(
        IActorRef commActor,
        int maxParallelism,
        TimeSpan? perInstanceTimeout = null,
        TimeSpan? reconcileTimeout = null)
    {
        var comms = new CommunicationService(
            Options.Create(new CommunicationOptions { DeploymentTimeout = TimeSpan.FromSeconds(30) }),
            NullLogger<CommunicationService>.Instance);
        comms.SetCommunicationActor(commActor);

        var options = Options.Create(new DeploymentManagerOptions
        {
            OperationLockTimeout = TimeSpan.FromSeconds(5),
            SiteDeploymentMaxParallelism = maxParallelism,
            SiteDeploymentTimeoutPerInstance = perInstanceTimeout ?? TimeSpan.FromSeconds(30),
            SiteDeploymentReconcileTimeout = reconcileTimeout ?? TimeSpan.FromSeconds(5)
        });

        return new DeploymentService(
            _repo, _siteRepo, _pipeline, comms, _lockManager, _audit,
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
            _staleProbe,
            options,
            Options.Create(new CommunicationOptions
            {
                CentralFetchBaseUrl = "https://central.test:9000",
                PendingDeploymentTtl = TimeSpan.FromMinutes(5)
            }),
            NullLogger<DeploymentService>.Instance);
    }

    [Fact]
    public async Task DeploySiteAsync_SlowInstance_OthersStillComplete_AndConcurrencyIsBounded()
    {
        const int instanceCount = 8;
        const int maxParallelism = 3;
        ArrangeInstances(instanceCount);

        var tracker = new ConcurrencyTracker();
        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(tracker, slowInstanceName: "Inst-01", slowDelay: TimeSpan.FromMilliseconds(400))));

        var service = CreateService(commActor, maxParallelism);

        var result = await service.DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        var summary = result.Value;

        // Every instance produced a row and all eight completed, including the
        // seven that were NOT waiting on the slow site round-trip.
        Assert.Equal(instanceCount, summary.InstanceResults.Count);
        Assert.Equal(instanceCount, summary.SuccessCount);
        Assert.Equal(0, summary.FailureCount);

        // The fan-out never exceeded the configured bound.
        Assert.True(tracker.MaxObserved <= maxParallelism,
            $"observed {tracker.MaxObserved} concurrent site round-trips, bound was {maxParallelism}");

        // ...and it genuinely WAS concurrent, so the assertion above is not
        // vacuously satisfied by a serial implementation.
        Assert.True(tracker.MaxObserved > 1, "site round-trips did not run concurrently at all");

        // Deployment identity: one distinct deployment id per instance.
        var deploymentIds = summary.InstanceResults.Select(r => r.DeploymentId).ToList();
        Assert.Equal(instanceCount, deploymentIds.Distinct().Count());
        Assert.DoesNotContain(deploymentIds, id => string.IsNullOrEmpty(id));
    }

    [Fact]
    public async Task DeploySiteAsync_WedgedInstance_TimesOutAlone_RestSucceed()
    {
        ArrangeInstances(4);

        var tracker = new ConcurrencyTracker();
        // Instance 2's round-trip is never answered — its per-instance deadline
        // must contain it rather than stalling the batch.
        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(tracker, slowInstanceName: "Inst-02", slowDelay: Timeout.InfiniteTimeSpan)));

        var service = CreateService(commActor, maxParallelism: 4, perInstanceTimeout: TimeSpan.FromMilliseconds(300));

        var result = await service.DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        var summary = result.Value;

        Assert.Equal(4, summary.InstanceResults.Count);
        Assert.Equal(3, summary.SuccessCount);
        Assert.Equal(1, summary.FailureCount);

        var wedged = Assert.Single(summary.InstanceResults, r => !r.Success);
        Assert.Equal("Inst-02", wedged.UniqueName);
    }

    [Fact]
    public async Task DeploySiteAsync_ReleasesEveryOperationLock()
    {
        ArrangeInstances(5);

        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(new ConcurrencyTracker(), slowInstanceName: null, slowDelay: TimeSpan.Zero)));

        var service = CreateService(commActor, maxParallelism: 2);

        await service.DeploySiteAsync(SiteId, "admin");

        // The per-instance operation lock is held from prepare through finalize;
        // once the batch is done every entry must be reclaimed, or a second bulk
        // deploy of the same site would deadlock on its own leftovers.
        Assert.Equal(0, _lockManager.TrackedLockCount);
    }

    [Fact]
    public async Task DeploySiteAsync_SiteWithNoInstances_SucceedsWithEmptySummary()
    {
        _siteRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any<CancellationToken>()).Returns([]);

        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(new ConcurrencyTracker(), slowInstanceName: null, slowDelay: TimeSpan.Zero)));

        var result = await CreateService(commActor, maxParallelism: 2).DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        Assert.Empty(result.Value.InstanceResults);
    }

    [Fact]
    public async Task DeploySiteAsync_UnknownSite_ReturnsFailure()
    {
        _siteRepo.GetSiteByIdAsync(99, Arg.Any<CancellationToken>()).Returns((Site?)null);

        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(new ConcurrencyTracker(), slowInstanceName: null, slowDelay: TimeSpan.Zero)));

        var result = await CreateService(commActor, maxParallelism: 2).DeploySiteAsync(99, "admin");

        Assert.True(result.IsFailure);
        Assert.Contains("not found", result.Error);
    }

    /// <summary>
    /// A <c>PendingDeployment</c>'s fetch token expires
    /// <c>PendingDeploymentTtl</c> after the row is STAGED, so any delay between
    /// staging and sending is dead time burned off the token's life. Staging the
    /// whole batch up front in phase 1 made that delay grow with batch size — the
    /// tail instances' tokens could expire before their command was ever sent, and
    /// the site's fetch then 404s.
    ///
    /// <para>
    /// This pins the fix: every instance's <c>RefreshDeploymentCommand</c> must
    /// arrive at the site carrying a FRESH <c>Timestamp</c> (the staging instant),
    /// regardless of how long the batch ahead of it took. Run serially with a slow
    /// site so the batch takes far longer than the freshness bound being asserted —
    /// under the old shape the last instance's token would already be ~2 s old.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_StagesEachTokenImmediatelyBeforeItsOwnSend()
    {
        const int instanceCount = 20;
        var perSendDelay = TimeSpan.FromMilliseconds(100);
        ArrangeInstances(instanceCount);

        var ages = new ConcurrentBag<TimeSpan>();
        var commActor = Sys.ActorOf(Props.Create(() => new TokenAgeRecordingSiteActor(ages, perSendDelay)));

        // Parallelism 1 makes the batch strictly serial, so the accumulated lag a
        // front-loaded staging phase would produce is at its largest.
        var service = CreateService(commActor, maxParallelism: 1);

        var sw = Stopwatch.StartNew();
        var result = await service.DeploySiteAsync(SiteId, "admin");
        sw.Stop();

        Assert.True(result.IsSuccess);
        Assert.Equal(instanceCount, result.Value.SuccessCount);
        Assert.Equal(instanceCount, ages.Count);

        // The batch really did take a long time — otherwise the freshness assertion
        // below would pass vacuously.
        Assert.True(sw.Elapsed > TimeSpan.FromMilliseconds(1500),
            $"the batch completed in {sw.ElapsedMilliseconds} ms; too fast to prove token freshness");

        // ...yet no token was stale when its command reached the site.
        var oldest = ages.Max();
        Assert.True(oldest < TimeSpan.FromMilliseconds(500),
            $"a fetch token was already {oldest.TotalMilliseconds:F0} ms old on arrival — " +
            "staging is not tracking the send.");
    }

    /// <summary>
    /// Cancelling mid-batch must not leak operation locks. Phase 1's
    /// <c>ThrowIfCancellationRequested</c> escapes <c>DeploySiteAsync</c> while every
    /// already-prepared deployment still holds its per-instance lock — and
    /// <c>OperationLockManager</c> hands out real semaphores, so an undisposed handle
    /// wedges that instance against every future mutating command for the life of
    /// the process. Their records must also be finalised as Failed rather than left
    /// InProgress.
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_CancelledMidPrepare_ReleasesEveryLock_AndFailsPreparedRecords()
    {
        ArrangeInstances(6);

        using var cts = new CancellationTokenSource();

        // Cancel while preparing the third instance: the first two are fully
        // prepared and holding locks when the loop's next cancellation check throws.
        _pipeline
            .FlattenAndValidateAsync(3, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
            .Returns(_ =>
            {
                cts.Cancel();
                var config = new FlattenedConfiguration { InstanceUniqueName = "Inst-03" };
                return Result<FlatteningPipelineResult>.Success(
                    new FlatteningPipelineResult(config, "sha256:3", ValidationResult.Success()));
            });

        var tracker = new ConcurrencyTracker();
        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(tracker, slowInstanceName: null, slowDelay: TimeSpan.Zero)));

        var service = CreateService(commActor, maxParallelism: 2);

        var result = await service.DeploySiteAsync(SiteId, "admin", cancellationToken: cts.Token);
        Assert.True(result.IsSuccess);
        Assert.Equal(6, result.Value.FailureCount);
        Assert.Equal(0, result.Value.SuccessCount);

        Assert.Equal(0, _lockManager.TrackedLockCount);
        Assert.Equal(0, tracker.MaxObserved); // No command may dispatch after the preparation deadline.

        // No prepared deployment may be left InProgress.
        await _repo.Received().UpdateDeploymentRecordAsync(
            Arg.Is<DeploymentRecord>(r => r.Status == DeploymentStatus.Failed),
            Arg.Any<CancellationToken>());
    }

    /// <summary>
    /// P7-2: a bulk site deploy must not restart deliberately-stopped equipment.
    ///
    /// <para>
    /// Deploy is a legal transition out of <c>Disabled</c> and the deploy path
    /// forces the instance to <c>Enabled</c> on success, so a site-wide push over an
    /// unfiltered instance list silently re-enabled every instance an operator had
    /// disabled. Bulk deploy therefore skips them outright — no flatten, no send, no
    /// state change — and reports them so the operator can see what was left alone.
    /// The single-instance path is unchanged: that is the explicit "yes, re-enable
    /// this one" action.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_SkipsDisabledInstances_WithoutDeployingOrReEnablingThem()
    {
        var instances = ArrangeInstances(5, disabledIds: [2, 4]);

        var commActor = Sys.ActorOf(Props.Create(() =>
            new ThrottledSiteActor(new ConcurrencyTracker(), slowInstanceName: null, slowDelay: TimeSpan.Zero)));

        var result = await CreateService(commActor, maxParallelism: 3).DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        var summary = result.Value;

        // Only the three Enabled/NotDeployed instances were attempted.
        Assert.Equal(3, summary.InstanceResults.Count);
        Assert.Equal(3, summary.SuccessCount);
        Assert.Equal(0, summary.FailureCount);
        Assert.DoesNotContain(summary.InstanceResults, r => r.UniqueName is "Inst-02" or "Inst-04");

        // The Disabled pair is reported as skipped, with the reason the operator needs.
        Assert.NotNull(summary.SkippedInstances);
        var skipped = summary.SkippedInstances;
        Assert.Equal(2, skipped.Count);
        Assert.Equal(
            new[] { "Inst-02", "Inst-04" },
            skipped.Select(s => s.UniqueName).Order().ToArray());
        Assert.All(skipped, s => Assert.Contains("deploy individually to re-enable", s.Reason));

        // ...and left completely untouched: still Disabled, and never handed to the
        // repository at all (no deployment record, no instance update, no snapshot).
        foreach (var id in new[] { 2, 4 })
        {
            Assert.Equal(InstanceState.Disabled, instances.Single(i => i.Id == id).State);
            await _pipeline.DidNotReceive().FlattenAndValidateAsync(
                id, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>());
        }

        await _repo.DidNotReceive().UpdateInstanceAsync(
            Arg.Is<Instance>(i => i.Id == 2 || i.Id == 4), Arg.Any<CancellationToken>());
        await _repo.DidNotReceive().AddDeploymentRecordAsync(
            Arg.Is<DeploymentRecord>(r => r.InstanceId == 2 || r.InstanceId == 4), Arg.Any<CancellationToken>());
    }

    /// <summary>
    /// P7-3: phase 1 is serial, so an unreachable site used to cost the batch one
    /// full 30-second <c>QueryTimeout</c> PER INSTANCE before a single deploy was
    /// sent — hours for a large site, so the CLI's 5-minute bulk deadline cancelled
    /// and the next attempt re-armed the same cost. Two bounds close it: a short
    /// per-reconcile deadline, and a once-per-batch circuit-breaker that skips the
    /// remaining reconciles after the first timeout.
    ///
    /// <para>
    /// The site here answers the deploy but NEVER answers the reconcile query, so
    /// the assertion that exactly ONE query was issued for six instances is the
    /// whole finding: without the breaker there would be six.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_UnreachableSite_ReconcilesOnceThenSkipsTheRestOfTheBatch()
    {
        const int instanceCount = 6;
        ArrangeInstances(instanceCount);

        // Reconciliation only runs when a prior record is stuck InProgress.
        for (var i = 1; i <= instanceCount; i++)
        {
            _repo.GetCurrentDeploymentStatusAsync(i, Arg.Any<CancellationToken>())
                .Returns(new DeploymentRecord($"prior-{i}", "admin")
                {
                    InstanceId = i,
                    Status = DeploymentStatus.InProgress,
                    RevisionHash = "sha256:old"
                });
        }

        var queryCount = 0;
        var commActor = Sys.ActorOf(Props.Create(() =>
            new SilentQuerySiteActor(() => Interlocked.Increment(ref queryCount))));

        var reconcileTimeout = TimeSpan.FromMilliseconds(300);
        var service = CreateService(commActor, maxParallelism: 3, reconcileTimeout: reconcileTimeout);

        var sw = Stopwatch.StartNew();
        var result = await service.DeploySiteAsync(SiteId, "admin");
        sw.Stop();

        Assert.True(result.IsSuccess);

        // The circuit-breaker opened after the first unanswered reconcile.
        Assert.Equal(1, Volatile.Read(ref queryCount));

        // Prepare therefore cost ONE reconcile deadline, not six.
        Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2),
            $"bulk prepare took {sw.ElapsedMilliseconds} ms — the per-batch reconcile bound is not holding");

        // A skipped reconcile is not a failed deploy: every instance still went out.
        Assert.Equal(instanceCount, result.Value.SuccessCount);
        Assert.Equal(0, result.Value.FailureCount);
    }

    /// <summary>
    /// Marks an instance as one a bulk deploy must treat as up to date: it is
    /// <c>Enabled</c>, central holds a deployed snapshot for it, and the staleness
    /// probe recomputes the very hash that snapshot recorded.
    /// </summary>
    /// <param name="instance">The instance to make up-to-date.</param>
    private void ArrangeUpToDate(Instance instance)
    {
        var hash = $"sha256:{instance.Id}";
        instance.State = InstanceState.Enabled;

        _repo.GetDeployedSnapshotByInstanceIdAsync(instance.Id, Arg.Any<CancellationToken>())
            .Returns(new DeployedConfigSnapshot($"prior-{instance.Id}", hash, "{}") { InstanceId = instance.Id });
        _repo.GetCurrentDeploymentStatusAsync(instance.Id, Arg.Any<CancellationToken>())
            .Returns(new DeploymentRecord($"prior-{instance.Id}", "admin")
            {
                InstanceId = instance.Id,
                Status = DeploymentStatus.Success,
                RevisionHash = hash,
            });
        _staleProbe.GetCurrentRevisionHashAsync(instance.Id, Arg.Any<CancellationToken>())
            .Returns(hash);
    }

    /// <summary>
    /// P7-6: a bulk site deploy must not redeploy instances whose configuration has
    /// not changed.
    ///
    /// <para>
    /// A redeploy is NOT a no-op at the site: applying a config RESETS that
    /// instance's static attribute overrides and CLEARS its native alarm state,
    /// both by design. Pushing one site-wide therefore destroyed live operational
    /// state on every instance in the batch, including the majority whose config
    /// the operator had not touched at all.
    /// </para>
    ///
    /// <para>
    /// The central-side proof that the site state survives is that the site is
    /// never told to apply anything: no <c>RefreshDeploymentCommand</c> is sent
    /// for a skipped instance, so nothing at the site runs the reset. The
    /// assertions below therefore pin the whole chain — no flatten, no deployment
    /// record, no staged config, no command — for the unchanged pair.
    /// </para>
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_SkipsUpToDateInstances_SoTheirSiteStateIsNeverReset()
    {
        var instances = ArrangeInstances(4);
        ArrangeUpToDate(instances.Single(i => i.Id == 2));
        ArrangeUpToDate(instances.Single(i => i.Id == 3));

        var applied = new ConcurrentBag<string>();
        var commActor = Sys.ActorOf(Props.Create(() => new RecordingSiteActor(applied)));

        var result = await CreateService(commActor, maxParallelism: 4).DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        var summary = result.Value;

        // Only the two genuinely-stale instances were deployed.
        Assert.Equal(2, summary.InstanceResults.Count);
        Assert.Equal(2, summary.SuccessCount);
        Assert.Equal(0, summary.FailureCount);
        Assert.Equal(new[] { "Inst-01", "Inst-04" }, summary.InstanceResults.Select(r => r.UniqueName).Order().ToArray());

        // The unchanged pair is reported as skipped, with the reason the operator needs.
        Assert.NotNull(summary.SkippedInstances);
        var skipped = summary.SkippedInstances;
        Assert.Equal(new[] { "Inst-02", "Inst-03" }, skipped.Select(s => s.UniqueName).Order().ToArray());
        Assert.All(skipped, s => Assert.Contains("up-to-date", s.Reason));

        // ...and nothing reached the site for them, which is what leaves their
        // static overrides and native alarm state alone.
        Assert.Equal(new[] { "Inst-01", "Inst-04" }, applied.Order().ToArray());

        foreach (var id in new[] { 2, 3 })
        {
            await _pipeline.DidNotReceive().FlattenAndValidateAsync(
                id, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>());
            await _repo.DidNotReceive().AddDeploymentRecordAsync(
                Arg.Is<DeploymentRecord>(r => r.InstanceId == id), Arg.Any<CancellationToken>());
            await _repo.DidNotReceive().AddPendingDeploymentAsync(
                Arg.Is<PendingDeployment>(p => p.InstanceId == id), Arg.Any<CancellationToken>());
        }
    }

    /// <summary>
    /// P7-6: <c>force</c> restores the unconditional behaviour for operators whose
    /// site state central cannot see — a restored data volume or a rebuilt node
    /// looks perfectly up to date from central and still needs the push.
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_Force_RedeploysUpToDateInstances()
    {
        var instances = ArrangeInstances(3);
        ArrangeUpToDate(instances.Single(i => i.Id == 2));

        var applied = new ConcurrentBag<string>();
        var commActor = Sys.ActorOf(Props.Create(() => new RecordingSiteActor(applied)));

        var result = await CreateService(commActor, maxParallelism: 3)
            .DeploySiteAsync(SiteId, "admin", force: true);

        Assert.True(result.IsSuccess);
        Assert.Equal(3, result.Value.SuccessCount);
        Assert.Empty(result.Value.SkippedInstances!);
        Assert.Contains("Inst-02", applied);
    }

    /// <summary>
    /// P7-6 composed with P7-2: <c>force</c> suppresses the up-to-date filter and
    /// ONLY that filter. A Disabled instance is never deployed at bulk scope,
    /// because forcing a bulk push is an operator saying "the site's config may be
    /// wrong", not "restart the equipment I stopped" — the single-instance deploy
    /// remains the explicit re-enable action.
    /// </summary>
    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task DeploySiteAsync_DisabledInstancesAreSkipped_RegardlessOfForce(bool force)
    {
        var instances = ArrangeInstances(3, disabledIds: [2]);

        var applied = new ConcurrentBag<string>();
        var commActor = Sys.ActorOf(Props.Create(() => new RecordingSiteActor(applied)));

        var result = await CreateService(commActor, maxParallelism: 3)
            .DeploySiteAsync(SiteId, "admin", force: force);

        Assert.True(result.IsSuccess);
        Assert.DoesNotContain("Inst-02", applied);
        Assert.Equal(InstanceState.Disabled, instances.Single(i => i.Id == 2).State);

        var skipped = Assert.Single(result.Value.SkippedInstances!);
        Assert.Equal("Inst-02", skipped.UniqueName);
        Assert.Contains("deploy individually to re-enable", skipped.Reason);
    }

    /// <summary>
    /// The up-to-date filter is deliberately conservative: every uncertainty
    /// resolves to "deploy it". These are the three cases where a matching hash is
    /// NOT sufficient — each one would be a silently skipped instance that actually
    /// needed the deploy.
    /// </summary>
    [Fact]
    public async Task DeploySiteAsync_MatchingHashIsNotEnough_WhenTheInstanceStillNeedsDeploying()
    {
        var instances = ArrangeInstances(3);

        // 1: NotDeployed. A leftover snapshot says nothing about an instance that
        // is not running — it must be deployed to become Enabled.
        ArrangeUpToDate(instances.Single(i => i.Id == 1));
        instances.Single(i => i.Id == 1).State = InstanceState.NotDeployed;

        // 2: prior record stuck InProgress. Prepare is where query-before-redeploy
        // reconciliation HEALS such a record; skipping here strands it forever.
        ArrangeUpToDate(instances.Single(i => i.Id == 2));
        _repo.GetCurrentDeploymentStatusAsync(2, Arg.Any<CancellationToken>())
            .Returns(new DeploymentRecord("prior-2", "admin")
            {
                InstanceId = 2,
                Status = DeploymentStatus.InProgress,
                RevisionHash = "sha256:2",
            });

        // 3: staleness indeterminate — the probe cannot compute a hash.
        ArrangeUpToDate(instances.Single(i => i.Id == 3));
        _staleProbe.GetCurrentRevisionHashAsync(3, Arg.Any<CancellationToken>()).Returns((string?)null);

        var applied = new ConcurrentBag<string>();
        var commActor = Sys.ActorOf(Props.Create(() => new RecordingSiteActor(applied)));

        var result = await CreateService(commActor, maxParallelism: 3).DeploySiteAsync(SiteId, "admin");

        Assert.True(result.IsSuccess);
        Assert.Empty(result.Value.SkippedInstances!);
        Assert.Equal(new[] { "Inst-01", "Inst-02", "Inst-03" }, applied.Order().ToArray());
    }

    /// <summary>Answers every deploy with Success, recording which instances the site was asked to apply.</summary>
    private sealed class RecordingSiteActor : ReceiveActor
    {
        public RecordingSiteActor(ConcurrentBag<string> applied)
        {
            Receive<SiteEnvelope>(env =>
            {
                if (env.Message is not RefreshDeploymentCommand cmd)
                    return;

                applied.Add(cmd.InstanceUniqueName);
                Sender.Tell(new DeploymentStatusResponse(
                    cmd.DeploymentId, cmd.InstanceUniqueName,
                    DeploymentStatus.Success, null, DateTimeOffset.UtcNow));
            });
        }
    }

    /// <summary>
    /// Answers <c>RefreshDeploymentCommand</c> normally but NEVER answers a
    /// <c>DeploymentStateQueryRequest</c>, counting how many arrived — a site that
    /// is up for deploys yet unreachable for the reconcile query, which is the shape
    /// a wedged or overloaded site presents.
    /// </summary>
    private sealed class SilentQuerySiteActor : ReceiveActor
    {
        public SilentQuerySiteActor(Action onQuery)
        {
            Receive<SiteEnvelope>(env =>
            {
                switch (env.Message)
                {
                    case DeploymentStateQueryRequest:
                        onQuery();
                        return;

                    case RefreshDeploymentCommand cmd:
                        Sender.Tell(new DeploymentStatusResponse(
                            cmd.DeploymentId, cmd.InstanceUniqueName,
                            DeploymentStatus.Success, null, DateTimeOffset.UtcNow));
                        return;
                }
            });
        }
    }

    /// <summary>
    /// Answers every <c>RefreshDeploymentCommand</c> after a fixed delay, recording
    /// how old each command's staging <c>Timestamp</c> already was on arrival.
    /// </summary>
    private sealed class TokenAgeRecordingSiteActor : ReceiveActor
    {
        public TokenAgeRecordingSiteActor(ConcurrentBag<TimeSpan> ages, TimeSpan delay)
        {
            Receive<SiteEnvelope>(env =>
            {
                if (env.Message is not RefreshDeploymentCommand cmd)
                    return;

                ages.Add(DateTimeOffset.UtcNow - cmd.Timestamp);

                var replyTo = Sender;
                Context.System.Scheduler.Advanced.ScheduleOnce(delay, () =>
                    replyTo.Tell(new DeploymentStatusResponse(
                        cmd.DeploymentId, cmd.InstanceUniqueName,
                        DeploymentStatus.Success, null, DateTimeOffset.UtcNow)));
            });
        }
    }

    /// <summary>Records the peak number of simultaneously in-flight site round-trips.</summary>
    private sealed class ConcurrencyTracker
    {
        private int _current;
        private int _max;

        /// <summary>Highest simultaneous in-flight count observed.</summary>
        public int MaxObserved => Volatile.Read(ref _max);

        /// <summary>Marks one round-trip as started and updates the peak.</summary>
        public void Enter()
        {
            var now = Interlocked.Increment(ref _current);
            int observed;
            while (now > (observed = Volatile.Read(ref _max)))
            {
                if (Interlocked.CompareExchange(ref _max, now, observed) == observed)
                    break;
            }
        }

        /// <summary>Marks one round-trip as finished.</summary>
        public void Exit() => Interlocked.Decrement(ref _current);
    }

    /// <summary>
    /// Stand-in site that answers <c>RefreshDeploymentCommand</c> with Success,
    /// optionally delaying (or never answering) one named instance so the test can
    /// observe the fan-out's bound and its per-instance deadline.
    ///
    /// <para>
    /// The reply is scheduled rather than sent inline, so the actor's mailbox is
    /// not the thing serialising the batch — otherwise the concurrency the test is
    /// measuring would be an artefact of the harness.
    /// </para>
    /// </summary>
    private sealed class ThrottledSiteActor : ReceiveActor
    {
        public ThrottledSiteActor(ConcurrencyTracker tracker, string? slowInstanceName, TimeSpan slowDelay)
        {
            Receive<SiteEnvelope>(env =>
            {
                if (env.Message is not RefreshDeploymentCommand cmd)
                    return;

                var replyTo = Sender;
                var isSlow = slowInstanceName != null && cmd.InstanceUniqueName == slowInstanceName;

                if (isSlow && slowDelay == Timeout.InfiniteTimeSpan)
                {
                    // Never answer: the caller's per-instance deadline must fire.
                    tracker.Enter();
                    return;
                }

                var delay = isSlow ? slowDelay : TimeSpan.FromMilliseconds(60);
                tracker.Enter();
                Context.System.Scheduler.Advanced.ScheduleOnce(delay, () =>
                {
                    tracker.Exit();
                    replyTo.Tell(new DeploymentStatusResponse(
                        cmd.DeploymentId, cmd.InstanceUniqueName,
                        DeploymentStatus.Success, null, DateTimeOffset.UtcNow));
                });
            });
        }
    }
}