feat(deploy): batch instance staleness over one flatten session

DeploymentService.GetInstanceStalenessAsync shares a FlattenSession across
the batch, reads only the (InstanceId, RevisionHash) projection and pays the
snapshot read + I-1 List normalisation only for a mismatch, so the fleet
aggregate agrees exactly with the per-instance Topology badge. The I-1 logic
is extracted from GetDeploymentComparisonAsync unchanged.

Claude-Session: https://claude.ai/code/session_01Y5ef34iUUMjmwnMpL9xgmB

src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs

@@ -1707,40 +1707,11 @@ public class DeploymentService
         // A snapshot that cannot be deserialized (corrupt /
         // older schema) still yields the frozen-hash staleness result, with a
         // null diff.
-        var deployedRevisionHash = snapshot.RevisionHash;
+        var deployedRevisionHash = ComputeDeployedHash(snapshot, instanceId, out var deployedConfig);
-        ConfigurationDiff? diff = null;
+        var diff = deployedConfig == null
-        try
+            ? null
-        {
+            : _diffService.ComputeDiff(
-            var deployedConfig = JsonSerializer.Deserialize<FlattenedConfiguration>(snapshot.ConfigurationJson);
-            if (deployedConfig != null)
-            {
-                deployedConfig = NormalizeListAttributeValues(deployedConfig);
-
-                // Recompute the deployed-side hash from the normalized snapshot so
-                // an old-form List value is not flagged stale against the native
-                // re-flatten. For a faithfully-stored scalar-only snapshot this
-                // reproduces the frozen RevisionHash exactly, so behaviour is
-                // unchanged outside the List-normalization case.
-                deployedRevisionHash = _revisionHashService.ComputeHash(deployedConfig);
-
-                diff = _diffService.ComputeDiff(
                 deployedConfig, currentConfig, deployedRevisionHash, currentHash);
-            }
-            else
-            {
-                _logger.LogWarning(
-                    "Deployed snapshot for instance {InstanceId} deserialized to null; " +
-                    "returning hash-based comparison without a structured diff",
-                    instanceId);
-            }
-        }
-        catch (JsonException ex)
-        {
-            _logger.LogWarning(ex,
-                "Could not deserialize deployed snapshot for instance {InstanceId}; " +
-                "returning hash-based comparison without a structured diff",
-                instanceId);
-        }
 
         var isStale = deployedRevisionHash != currentHash;
 
@@ -1755,6 +1726,153 @@ public class DeploymentService
         return Result<DeploymentComparisonResult>.Success(result);
     }
 
+    /// <summary>
+    /// Derives the deployed-side staleness hash from a stored snapshot, applying
+    /// the I-1 List normalization: the snapshot's <c>ConfigurationJson</c> is
+    /// deserialized, its <see cref="DataType.List"/> values round-tripped to native
+    /// form, and the hash recomputed over the result — so a List attribute deployed
+    /// in the old quoted form is not flagged stale against a native re-flatten. For
+    /// a faithfully-stored scalar-only snapshot this reproduces the frozen
+    /// <see cref="DeployedConfigSnapshot.RevisionHash"/> exactly.
+    ///
+    /// <para>
+    /// A snapshot that cannot be deserialized (corrupt / older schema) falls back to
+    /// the frozen hash and yields a <see langword="null"/> normalized config, which
+    /// the comparison path renders as "no structured diff". Shared by
+    /// <see cref="GetDeploymentComparisonAsync"/> and
+    /// <see cref="GetInstanceStalenessAsync"/> so the per-instance badge and the
+    /// fleet aggregate can never disagree.
+    /// </para>
+    /// </summary>
+    /// <param name="snapshot">The stored deployed-config snapshot.</param>
+    /// <param name="instanceId">Instance the snapshot belongs to; named in the warning logs.</param>
+    /// <param name="normalizedDeployedConfig">
+    /// The normalized deployed configuration, or <see langword="null"/> when the
+    /// snapshot could not be deserialized.
+    /// </param>
+    /// <returns>The hash to compare the current re-flatten's hash against.</returns>
+    private string ComputeDeployedHash(
+        DeployedConfigSnapshot snapshot,
+        int instanceId,
+        out FlattenedConfiguration? normalizedDeployedConfig)
+    {
+        normalizedDeployedConfig = null;
+        try
+        {
+            var deployedConfig = JsonSerializer.Deserialize<FlattenedConfiguration>(snapshot.ConfigurationJson);
+            if (deployedConfig != null)
+            {
+                normalizedDeployedConfig = NormalizeListAttributeValues(deployedConfig);
+                return _revisionHashService.ComputeHash(normalizedDeployedConfig);
+            }
+
+            _logger.LogWarning(
+                "Deployed snapshot for instance {InstanceId} deserialized to null; " +
+                "returning hash-based comparison without a structured diff",
+                instanceId);
+        }
+        catch (JsonException ex)
+        {
+            _logger.LogWarning(ex,
+                "Could not deserialize deployed snapshot for instance {InstanceId}; " +
+                "returning hash-based comparison without a structured diff",
+                instanceId);
+        }
+
+        return snapshot.RevisionHash;
+    }
+
+    /// <summary>
+    /// Evaluates staleness for a batch of instances — the fleet-wide read behind
+    /// the pending-updates aggregate — agreeing exactly with the per-instance
+    /// Topology badge (<see cref="GetDeploymentComparisonAsync"/>) while costing a
+    /// fraction of running that method N times.
+    ///
+    /// <para>
+    /// <b>How it stays cheap.</b> All the flattens share ONE
+    /// <see cref="FlattenSession"/>, so a template chain used by N instances is
+    /// walked once (the same trick <see cref="DeploySiteAsync"/> uses). The deployed
+    /// side is read as the <c>(InstanceId, RevisionHash)</c> projection
+    /// (<see cref="IDeploymentManagerRepository.GetDeployedRevisionHashesAsync"/>),
+    /// never the fat <c>ConfigurationJson</c> for the whole fleet. Only when that
+    /// projected hash DISAGREES with the re-flatten is the snapshot read and the
+    /// I-1 List normalization applied — which is what makes the aggregate agree with
+    /// the badge on an old-form List snapshot without paying for it fleet-wide.
+    /// </para>
+    ///
+    /// <para>
+    /// <b>Sequential by necessity.</b> The repository's <c>DbContext</c> and the
+    /// session are both single-threaded (see <see cref="FlattenSession"/>
+    /// "Threading"), so the loop must not be parallelised.
+    /// </para>
+    ///
+    /// <para>
+    /// <b>Result shape.</b> An instance with no deployed snapshot is OMITTED — it
+    /// has nothing to be stale against, and a snapshot that vanished between the
+    /// projection read and the snapshot read means it was undeployed concurrently,
+    /// which is treated the same way. <see langword="null"/> means "could not
+    /// evaluate" (its flatten failed): unknown, never counted as stale.
+    /// </para>
+    /// </summary>
+    /// <param name="instanceIds">Instances to evaluate; duplicates are collapsed.</param>
+    /// <param name="cancellationToken">Cancellation token, honoured between instances.</param>
+    /// <returns>
+    /// A map of instance id → <see langword="true"/> (stale), <see langword="false"/>
+    /// (up to date) or <see langword="null"/> (could not evaluate), omitting
+    /// instances with no deployed snapshot.
+    /// </returns>
+    public async Task<IReadOnlyDictionary<int, bool?>> GetInstanceStalenessAsync(
+        IReadOnlyCollection<int> instanceIds,
+        CancellationToken cancellationToken = default)
+    {
+        var staleness = new Dictionary<int, bool?>();
+        if (instanceIds.Count == 0)
+            return staleness;
+
+        var deployedHashes = await _repository.GetDeployedRevisionHashesAsync(cancellationToken);
+        var session = _flatteningPipeline.CreateSession();
+
+        foreach (var instanceId in instanceIds.Distinct())
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (!deployedHashes.TryGetValue(instanceId, out var deployedHash))
+                continue;
+
+            // Read-only path: hash only, so skip the Roslyn compile — the deploy
+            // gate remains the authoritative compile.
+            var currentResult = await _flatteningPipeline.FlattenAndValidateAsync(
+                instanceId, cancellationToken, validateScripts: false, session);
+            if (currentResult.IsFailure)
+            {
+                _logger.LogWarning(
+                    "Could not evaluate staleness of instance {InstanceId}: {Error}",
+                    instanceId, currentResult.Error);
+                staleness[instanceId] = null;
+                continue;
+            }
+
+            var currentHash = currentResult.Value.RevisionHash;
+            if (string.Equals(currentHash, deployedHash, StringComparison.Ordinal))
+            {
+                staleness[instanceId] = false;
+                continue;
+            }
+
+            // Mismatch only: pay the fat snapshot read + I-1 normalization, so an
+            // old-form List snapshot reads exactly as the Topology badge renders it.
+            var snapshot = await _repository.GetDeployedSnapshotByInstanceIdAsync(instanceId, cancellationToken);
+            if (snapshot == null)
+                continue;
+
+            var normalizedDeployedHash = ComputeDeployedHash(snapshot, instanceId, out _);
+            staleness[instanceId] =
+                !string.Equals(normalizedDeployedHash, currentHash, StringComparison.Ordinal);
+        }
+
+        return staleness;
+    }
+
     /// <summary>
     /// I-1 (latent): returns a copy of <paramref name="config"/> whose
     /// <see cref="DataType.List"/> attribute values have been round-tripped through

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

@@ -0,0 +1,273 @@
+using System.Text.Json;
+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.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;
+using ZB.MOM.WW.ScadaBridge.Communication;
+using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
+using ZB.MOM.WW.ScadaBridge.TemplateEngine.Validation;
+
+namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests;
+
+/// <summary>
+/// Coverage for <see cref="DeploymentService.GetInstanceStalenessAsync"/> — the
+/// fleet-wide batch behind the pending-updates aggregate.
+///
+/// <para>
+/// The three properties under test are the ones that make the batch both cheap
+/// and exactly consistent with the per-instance Topology badge: ONE shared
+/// <see cref="FlattenSession"/> for the whole batch, a hash-projection read for
+/// everyone, and the fat snapshot read + I-1 List normalisation paid ONLY when
+/// the projected hash already disagrees.
+/// </para>
+/// </summary>
+public class DeploymentServiceStalenessTests
+{
+    private readonly IDeploymentManagerRepository _repo = Substitute.For<IDeploymentManagerRepository>();
+    private readonly IFlatteningPipeline _pipeline = Substitute.For<IFlatteningPipeline>();
+    private readonly RevisionHashService _hasher = new();
+    private readonly DeploymentService _service;
+
+    /// <summary>The single session every flatten in a batch must share.</summary>
+    private readonly FlattenSession _session = new(new TemplateGraphWatermark());
+
+    /// <summary>
+    /// Builds the service over substituted repository and pipeline seams; every
+    /// other dependency is the real concrete collaborator (as elsewhere in this
+    /// suite) because the staleness path only exercises hashing.
+    /// </summary>
+    public DeploymentServiceStalenessTests()
+    {
+        _pipeline.CreateSession().Returns(_session);
+
+        var siteRepo = Substitute.For<ISiteRepository>();
+        siteRepo.GetSiteByIdAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo =>
+            {
+                var id = callInfo.ArgAt<int>(0);
+                return new Site($"Test Site {id}", $"site-{id}") { Id = id };
+            });
+
+        _service = new DeploymentService(
+            _repo,
+            siteRepo,
+            _pipeline,
+            new CommunicationService(
+                Options.Create(new CommunicationOptions()),
+                NullLogger<CommunicationService>.Instance),
+            new OperationLockManager(),
+            Substitute.For<IAuditService>(),
+            new DiffService(),
+            _hasher,
+            new DeploymentStatusNotifier(NullLogger<DeploymentStatusNotifier>.Instance),
+            Substitute.For<IStaleInstanceProbe>(),
+            Options.Create(new DeploymentManagerOptions
+            {
+                OperationLockTimeout = TimeSpan.FromSeconds(5)
+            }),
+            Options.Create(new CommunicationOptions()),
+            NullLogger<DeploymentService>.Instance);
+    }
+
+    /// <summary>Builds a one-attribute flattened config with the given scalar value.</summary>
+    private static FlattenedConfiguration ScalarConfig(string value) => new()
+    {
+        InstanceUniqueName = "TestInst",
+        Attributes = [new ResolvedAttribute { CanonicalName = "Setpoint", Value = value, DataType = "Int32" }]
+    };
+
+    /// <summary>Stubs a successful flatten of <paramref name="instanceId"/>.</summary>
+    private void ArrangeFlatten(int instanceId, FlattenedConfiguration config, string hash) =>
+        _pipeline.FlattenAndValidateAsync(
+                instanceId, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
+            .Returns(Result<FlatteningPipelineResult>.Success(
+                new FlatteningPipelineResult(config, hash, ValidationResult.Success())));
+
+    /// <summary>Stubs the fat snapshot read for <paramref name="instanceId"/>.</summary>
+    private void ArrangeSnapshot(int instanceId, FlattenedConfiguration config, string frozenHash) =>
+        _repo.GetDeployedSnapshotByInstanceIdAsync(instanceId, Arg.Any<CancellationToken>())
+            .Returns(new DeployedConfigSnapshot("dep1", frozenHash, JsonSerializer.Serialize(config))
+            {
+                InstanceId = instanceId,
+                DeployedAt = DateTimeOffset.UtcNow
+            });
+
+    /// <summary>Stubs the (InstanceId, RevisionHash) projection read.</summary>
+    private void ArrangeDeployedHashes(params (int InstanceId, string Hash)[] rows) =>
+        _repo.GetDeployedRevisionHashesAsync(Arg.Any<CancellationToken>())
+            .Returns(rows.ToDictionary(r => r.InstanceId, r => r.Hash));
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_SharesOneSessionAcrossTheBatch()
+    {
+        var config = ScalarConfig("5");
+        var hash = _hasher.ComputeHash(config);
+        ArrangeDeployedHashes((1, hash), (2, hash), (3, hash));
+        ArrangeFlatten(1, config, hash);
+        ArrangeFlatten(2, config, hash);
+        ArrangeFlatten(3, config, hash);
+
+        var result = await _service.GetInstanceStalenessAsync([1, 2, 3]);
+
+        Assert.Equal(3, result.Count);
+        // One session created, and the SAME instance handed to every flatten —
+        // that is what makes a template chain shared by N instances walked once.
+        _pipeline.Received(1).CreateSession();
+        await _pipeline.Received(1).FlattenAndValidateAsync(1, Arg.Any<CancellationToken>(), false, _session);
+        await _pipeline.Received(1).FlattenAndValidateAsync(2, Arg.Any<CancellationToken>(), false, _session);
+        await _pipeline.Received(1).FlattenAndValidateAsync(3, Arg.Any<CancellationToken>(), false, _session);
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_HashMatches_NotStaleAndNoSnapshotRead()
+    {
+        var config = ScalarConfig("5");
+        var hash = _hasher.ComputeHash(config);
+        ArrangeDeployedHashes((1, hash));
+        ArrangeFlatten(1, config, hash);
+
+        var result = await _service.GetInstanceStalenessAsync([1]);
+
+        Assert.False(result[1]);
+        // The whole point of the projection read: an up-to-date fleet never
+        // deserialises a single ConfigurationJson.
+        await _repo.DidNotReceive().GetDeployedSnapshotByInstanceIdAsync(1, Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_OldFormListSnapshot_NormalisesToNotStale()
+    {
+        // I-1 parity with the Topology badge: a List attribute deployed in the OLD
+        // quoted form froze a hash that differs from the native re-flatten's, so
+        // the projection read reports a mismatch. Re-reading the snapshot and
+        // recomputing over the normalised config collapses old form → native, and
+        // the instance is correctly NOT stale.
+        var oldFormConfig = new FlattenedConfiguration
+        {
+            InstanceUniqueName = "ListInst",
+            Attributes =
+            [
+                new ResolvedAttribute
+                {
+                    CanonicalName = "Thresholds",
+                    DataType = "List",
+                    ElementDataType = "Int32",
+                    Value = "[\"10\",\"20\",\"30\"]" // OLD quoted form
+                }
+            ]
+        };
+        var nativeConfig = oldFormConfig with
+        {
+            Attributes = [oldFormConfig.Attributes[0] with { Value = "[10,20,30]" }]
+        };
+
+        var frozenOldFormHash = _hasher.ComputeHash(oldFormConfig);
+        var nativeHash = _hasher.ComputeHash(nativeConfig);
+        // Sanity: without the normalisation this instance WOULD read as stale.
+        Assert.NotEqual(frozenOldFormHash, nativeHash);
+
+        ArrangeDeployedHashes((50, frozenOldFormHash));
+        ArrangeSnapshot(50, oldFormConfig, frozenOldFormHash);
+        ArrangeFlatten(50, nativeConfig, nativeHash);
+
+        var result = await _service.GetInstanceStalenessAsync([50]);
+
+        Assert.False(result[50]);
+        await _repo.Received(1).GetDeployedSnapshotByInstanceIdAsync(50, Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_GenuineMismatch_IsStale()
+    {
+        var deployedConfig = ScalarConfig("5");
+        var deployedHash = _hasher.ComputeHash(deployedConfig);
+        var currentConfig = ScalarConfig("9");
+        var currentHash = _hasher.ComputeHash(currentConfig);
+
+        ArrangeDeployedHashes((1, deployedHash));
+        ArrangeSnapshot(1, deployedConfig, deployedHash);
+        ArrangeFlatten(1, currentConfig, currentHash);
+
+        var result = await _service.GetInstanceStalenessAsync([1]);
+
+        Assert.True(result[1]);
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_FlattenFails_ReportsUnknown()
+    {
+        ArrangeDeployedHashes((1, "sha256:deployed"));
+        _pipeline.FlattenAndValidateAsync(
+                1, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
+            .Returns(Result<FlatteningPipelineResult>.Failure("template missing"));
+
+        var result = await _service.GetInstanceStalenessAsync([1]);
+
+        // null = could not evaluate. Never counted as stale.
+        Assert.True(result.ContainsKey(1));
+        Assert.Null(result[1]);
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_InstanceWithoutSnapshot_IsOmitted()
+    {
+        var config = ScalarConfig("5");
+        var hash = _hasher.ComputeHash(config);
+        ArrangeDeployedHashes((1, hash));
+        ArrangeFlatten(1, config, hash);
+
+        var result = await _service.GetInstanceStalenessAsync([1, 2]);
+
+        Assert.True(result.ContainsKey(1));
+        // Instance 2 has never been deployed — not stale, not unknown, absent.
+        Assert.False(result.ContainsKey(2));
+        await _pipeline.DidNotReceive().FlattenAndValidateAsync(
+            2, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>());
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_SnapshotVanishedBetweenReads_IsOmitted()
+    {
+        // Undeployed between the projection read and the mismatch's snapshot read:
+        // there is nothing left to be stale against, so report nothing at all.
+        ArrangeDeployedHashes((1, "sha256:deployed"));
+        _repo.GetDeployedSnapshotByInstanceIdAsync(1, Arg.Any<CancellationToken>())
+            .Returns((DeployedConfigSnapshot?)null);
+        ArrangeFlatten(1, ScalarConfig("5"), "sha256:current");
+
+        var result = await _service.GetInstanceStalenessAsync([1]);
+
+        Assert.False(result.ContainsKey(1));
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_DuplicateIds_FlattenedOnce()
+    {
+        var config = ScalarConfig("5");
+        var hash = _hasher.ComputeHash(config);
+        ArrangeDeployedHashes((1, hash));
+        ArrangeFlatten(1, config, hash);
+
+        var result = await _service.GetInstanceStalenessAsync([1, 1, 1]);
+
+        Assert.Single(result);
+        await _pipeline.Received(1).FlattenAndValidateAsync(
+            1, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>());
+    }
+
+    [Fact]
+    public async Task GetInstanceStalenessAsync_EmptyInput_ReturnsEmptyWithoutReads()
+    {
+        var result = await _service.GetInstanceStalenessAsync([]);
+
+        Assert.Empty(result);
+        await _repo.DidNotReceive().GetDeployedRevisionHashesAsync(Arg.Any<CancellationToken>());
+    }
+}