fix(deployment): bulk-deploy up-to-date pre-filter uses batch hash/status reads and the operation's shared flatten session; completion log reports pre-filter cost (review 2026-10-21 04/P1)

2026-10-22 06:46:03 -04:00
parent df67fabe68
commit 32e63eaa96
11 changed files with 247 additions and 78 deletions
@@ -10,7 +10,6 @@ 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;
@@ -35,7 +34,10 @@ public class DeploySiteAsyncTests : TestKit
    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>();
+
+    // Batch reads behind the up-to-date pre-filter (one query each per bulk deploy).
+    private readonly Dictionary<int, string> _deployedHashes = new();
+    private readonly Dictionary<int, DeploymentRecord> _currentRecords = new();
    private readonly OperationLockManager _lockManager = new();

    /// <summary>Initializes the test class with substituted repositories, a flattening pipeline and a stale-instance probe.</summary>
@@ -45,6 +47,13 @@ public class DeploySiteAsyncTests : TestKit
            .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()));
+
+        _repo.GetDeployedRevisionHashesAsync(Arg.Any<CancellationToken>())
+            .Returns(_ => (IReadOnlyDictionary<int, string>)new Dictionary<int, string>(_deployedHashes));
+        _repo.GetCurrentDeploymentStatusesAsync(Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>())
+            .Returns(ci => (IReadOnlyDictionary<int, DeploymentRecord>)ci.ArgAt<IReadOnlyCollection<int>>(0)
+                .Where(_currentRecords.ContainsKey)
+                .ToDictionary(id => id, id => _currentRecords[id]));
    }

    /// <summary>Seeds <paramref name="count"/> deployable instances at the test site.</summary>
@@ -102,7 +111,6 @@ public class DeploySiteAsyncTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            _staleProbe,
            options,
            Options.Create(new CommunicationOptions
            {
@@ -458,17 +466,17 @@ public class DeploySiteAsyncTests : TestKit
        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);
+        _deployedHashes[instance.Id] = hash;
+        var prior = new DeploymentRecord($"prior-{instance.Id}", "admin")
+        {
+            InstanceId = instance.Id,
+            Status = DeploymentStatus.Success,
+            RevisionHash = hash,
+        };
+        _currentRecords[instance.Id] = prior;
+        _repo.GetCurrentDeploymentStatusAsync(instance.Id, Arg.Any<CancellationToken>()).Returns(prior);
+        _pipeline.ComputeRevisionHashAsync(instance.Id, Arg.Any<FlattenSession?>(), Arg.Any<CancellationToken>())
+            .Returns(Result<string>.Success(hash));
    }

    /// <summary>
@@ -534,6 +542,56 @@ public class DeploySiteAsyncTests : TestKit
        }
    }

+    /// <summary>
+    /// review 2026-10-21 04/P1: the up-to-date pre-filter costs a fixed number of batch reads
+    /// (deployed-hash projection + one current-record query) — never a per-instance snapshot
+    /// load with its <c>ConfigurationJson</c> or a per-instance status query — and its
+    /// hash-only re-flattens run on the bulk operation's ONE shared session, the same one
+    /// phase 1 prepares on.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DeploySiteAsync_PreFilter_UsesBatchReadsAndTheSharedSession()
+    {
+        var instances = ArrangeInstances(4);
+        ArrangeUpToDate(instances.Single(i => i.Id == 2));
+        ArrangeUpToDate(instances.Single(i => i.Id == 3));
+        // Instance 4 is Enabled but its config changed: it is probed, found stale, deployed.
+        ArrangeUpToDate(instances.Single(i => i.Id == 4));
+        _deployedHashes[4] = "sha256:old-4";
+
+        var sessions = new List<FlattenSession?>();
+        _pipeline.When(p => p.ComputeRevisionHashAsync(
+                Arg.Any<int>(), Arg.Any<FlattenSession?>(), Arg.Any<CancellationToken>()))
+            .Do(ci => sessions.Add(ci.ArgAt<FlattenSession?>(1)));
+        _pipeline.When(p => p.FlattenAndValidateAsync(
+                Arg.Any<int>(), Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>()))
+            .Do(ci => sessions.Add(ci.ArgAt<FlattenSession?>(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);
+        Assert.Equal(new[] { "Inst-01", "Inst-04" }, applied.Order().ToArray());
+
+        await _repo.Received(1).GetDeployedRevisionHashesAsync(Arg.Any<CancellationToken>());
+        await _repo.Received(1).GetCurrentDeploymentStatusesAsync(
+            Arg.Is<IReadOnlyCollection<int>>(ids => ids.Order().SequenceEqual(new[] { 2, 3, 4 })),
+            Arg.Any<CancellationToken>());
+        foreach (var id in new[] { 2, 3 })
+        {
+            await _repo.DidNotReceive().GetDeployedSnapshotByInstanceIdAsync(id, Arg.Any<CancellationToken>());
+            await _repo.DidNotReceive().GetCurrentDeploymentStatusAsync(id, Arg.Any<CancellationToken>());
+        }
+
+        // One session for the whole operation: pre-filter probes and prepare flattens alike.
+        _pipeline.Received(1).CreateSession();
+        Assert.NotEmpty(sessions);
+        Assert.All(sessions, s => Assert.Same(sessions[0], s));
+    }
+
    /// <summary>
    /// <c>force</c> restores the unconditional behaviour for operators whose
    /// site state central cannot see — a restored data volume or a rebuilt node
@@ -609,17 +667,19 @@ public class DeploySiteAsyncTests : TestKit
        // 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",
-            });
+        var stuck = new DeploymentRecord("prior-2", "admin")
+        {
+            InstanceId = 2,
+            Status = DeploymentStatus.InProgress,
+            RevisionHash = "sha256:2",
+        };
+        _currentRecords[2] = stuck;
+        _repo.GetCurrentDeploymentStatusAsync(2, Arg.Any<CancellationToken>()).Returns(stuck);

-        // 3: staleness indeterminate — the probe cannot compute a hash.
+        // 3: staleness indeterminate — the hash-only flatten cannot compute a hash.
        ArrangeUpToDate(instances.Single(i => i.Id == 3));
-        _staleProbe.GetCurrentRevisionHashAsync(3, Arg.Any<CancellationToken>()).Returns((string?)null);
+        _pipeline.ComputeRevisionHashAsync(3, Arg.Any<FlattenSession?>(), Arg.Any<CancellationToken>())
+            .Returns(Result<string>.Failure("template chain unresolvable"));

        var applied = new ConcurrentBag<string>();
        var commActor = Sys.ActorOf(Props.Create(() => new RecordingSiteActor(applied)));
@@ -7,7 +7,6 @@ using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
 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.Types;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
@@ -71,7 +70,6 @@ public class DeploymentServiceStalenessTests
            new DiffService(),
            _hasher,
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5)
@@ -9,7 +9,6 @@ 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.Messages.Lifecycle;
 using ZB.MOM.WW.ScadaBridge.Commons.Observability;
@@ -56,7 +55,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            options,
            DeployCommOptions(),
            NullLogger<DeploymentService>.Instance);
@@ -202,7 +200,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions { OperationLockTimeout = TimeSpan.FromSeconds(5) }),
            DeployCommOptions(),
            NullLogger<DeploymentService>.Instance);
@@ -435,7 +432,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions { OperationLockTimeout = TimeSpan.FromSeconds(5) }),
            DeployCommOptions(),
            NullLogger<DeploymentService>.Instance);
@@ -1251,7 +1247,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions { OperationLockTimeout = TimeSpan.FromSeconds(5) }),
            DeployCommOptions(centralFetchBaseUrl),
            NullLogger<DeploymentService>.Instance);
@@ -1871,7 +1866,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5),
@@ -1932,7 +1926,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5),
@@ -1984,7 +1977,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5),
@@ -2033,7 +2025,6 @@ public class DeploymentServiceTests : TestKit
            new DiffService(),
            new RevisionHashService(),
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions
            {
                OperationLockTimeout = TimeSpan.FromSeconds(5),
@@ -7,7 +7,6 @@ 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.Types;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
@@ -63,7 +62,7 @@ public class DeploymentStatusNotifierTests : TestKit
        _service = new DeploymentService(
            _repo, siteRepo, _pipeline, _comms, _lockManager, _audit,
            new DiffService(), new RevisionHashService(), _notifier,
-            Substitute.For<IStaleInstanceProbe>(), options,
+            options,
            // Notify-and-fetch: a non-empty CentralFetchBaseUrl lets the
            // deploy path reach the send (and, here, the catch block) rather than
            // tripping the fail-fast configuration guard.
@@ -7,7 +7,6 @@ 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.Types;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
@@ -81,7 +80,6 @@ public class PendingUpdatesEvaluatorTests
            new DiffService(),
            _hasher,
            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
-            Substitute.For<IStaleInstanceProbe>(),
            Options.Create(new DeploymentManagerOptions()),
            Options.Create(new CommunicationOptions()),
            NullLogger<DeploymentService>.Instance);