From dc60ba1212973ce0fa67c8e5316d6444746fb80b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 14 Sep 2026 13:40:19 -0400 Subject: [PATCH] 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 --- .../DeploymentService.cs | 186 +++++++++--- .../DeploymentServiceStalenessTests.cs | 273 ++++++++++++++++++ 2 files changed, 425 insertions(+), 34 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceStalenessTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs index 00990633..47811131 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs +++ b/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; - ConfigurationDiff? diff = null; - try - { - var deployedConfig = JsonSerializer.Deserialize(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 deployedRevisionHash = ComputeDeployedHash(snapshot, instanceId, out var deployedConfig); + var diff = deployedConfig == null + ? null + : _diffService.ComputeDiff( + deployedConfig, currentConfig, deployedRevisionHash, currentHash); var isStale = deployedRevisionHash != currentHash; @@ -1755,6 +1726,153 @@ public class DeploymentService return Result.Success(result); } + /// + /// Derives the deployed-side staleness hash from a stored snapshot, applying + /// the I-1 List normalization: the snapshot's ConfigurationJson is + /// deserialized, its 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 + /// exactly. + /// + /// + /// A snapshot that cannot be deserialized (corrupt / older schema) falls back to + /// the frozen hash and yields a normalized config, which + /// the comparison path renders as "no structured diff". Shared by + /// and + /// so the per-instance badge and the + /// fleet aggregate can never disagree. + /// + /// + /// The stored deployed-config snapshot. + /// Instance the snapshot belongs to; named in the warning logs. + /// + /// The normalized deployed configuration, or when the + /// snapshot could not be deserialized. + /// + /// The hash to compare the current re-flatten's hash against. + private string ComputeDeployedHash( + DeployedConfigSnapshot snapshot, + int instanceId, + out FlattenedConfiguration? normalizedDeployedConfig) + { + normalizedDeployedConfig = null; + try + { + var deployedConfig = JsonSerializer.Deserialize(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; + } + + /// + /// Evaluates staleness for a batch of instances — the fleet-wide read behind + /// the pending-updates aggregate — agreeing exactly with the per-instance + /// Topology badge () while costing a + /// fraction of running that method N times. + /// + /// + /// How it stays cheap. All the flattens share ONE + /// , so a template chain used by N instances is + /// walked once (the same trick uses). The deployed + /// side is read as the (InstanceId, RevisionHash) projection + /// (), + /// never the fat ConfigurationJson 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. + /// + /// + /// + /// Sequential by necessity. The repository's DbContext and the + /// session are both single-threaded (see + /// "Threading"), so the loop must not be parallelised. + /// + /// + /// + /// Result shape. 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. means "could not + /// evaluate" (its flatten failed): unknown, never counted as stale. + /// + /// + /// Instances to evaluate; duplicates are collapsed. + /// Cancellation token, honoured between instances. + /// + /// A map of instance id → (stale), + /// (up to date) or (could not evaluate), omitting + /// instances with no deployed snapshot. + /// + public async Task> GetInstanceStalenessAsync( + IReadOnlyCollection instanceIds, + CancellationToken cancellationToken = default) + { + var staleness = new Dictionary(); + 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; + } + /// /// I-1 (latent): returns a copy of whose /// attribute values have been round-tripped through diff --git a/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceStalenessTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceStalenessTests.cs new file mode 100644 index 00000000..002991c0 --- /dev/null +++ b/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; + +/// +/// Coverage for — the +/// fleet-wide batch behind the pending-updates aggregate. +/// +/// +/// 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 +/// 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. +/// +/// +public class DeploymentServiceStalenessTests +{ + private readonly IDeploymentManagerRepository _repo = Substitute.For(); + private readonly IFlatteningPipeline _pipeline = Substitute.For(); + private readonly RevisionHashService _hasher = new(); + private readonly DeploymentService _service; + + /// The single session every flatten in a batch must share. + private readonly FlattenSession _session = new(new TemplateGraphWatermark()); + + /// + /// 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. + /// + public DeploymentServiceStalenessTests() + { + _pipeline.CreateSession().Returns(_session); + + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(Arg.Any(), Arg.Any()) + .Returns(callInfo => + { + var id = callInfo.ArgAt(0); + return new Site($"Test Site {id}", $"site-{id}") { Id = id }; + }); + + _service = new DeploymentService( + _repo, + siteRepo, + _pipeline, + new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance), + new OperationLockManager(), + Substitute.For(), + new DiffService(), + _hasher, + new DeploymentStatusNotifier(NullLogger.Instance), + Substitute.For(), + Options.Create(new DeploymentManagerOptions + { + OperationLockTimeout = TimeSpan.FromSeconds(5) + }), + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + } + + /// Builds a one-attribute flattened config with the given scalar value. + private static FlattenedConfiguration ScalarConfig(string value) => new() + { + InstanceUniqueName = "TestInst", + Attributes = [new ResolvedAttribute { CanonicalName = "Setpoint", Value = value, DataType = "Int32" }] + }; + + /// Stubs a successful flatten of . + private void ArrangeFlatten(int instanceId, FlattenedConfiguration config, string hash) => + _pipeline.FlattenAndValidateAsync( + instanceId, Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Result.Success( + new FlatteningPipelineResult(config, hash, ValidationResult.Success()))); + + /// Stubs the fat snapshot read for . + private void ArrangeSnapshot(int instanceId, FlattenedConfiguration config, string frozenHash) => + _repo.GetDeployedSnapshotByInstanceIdAsync(instanceId, Arg.Any()) + .Returns(new DeployedConfigSnapshot("dep1", frozenHash, JsonSerializer.Serialize(config)) + { + InstanceId = instanceId, + DeployedAt = DateTimeOffset.UtcNow + }); + + /// Stubs the (InstanceId, RevisionHash) projection read. + private void ArrangeDeployedHashes(params (int InstanceId, string Hash)[] rows) => + _repo.GetDeployedRevisionHashesAsync(Arg.Any()) + .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(), false, _session); + await _pipeline.Received(1).FlattenAndValidateAsync(2, Arg.Any(), false, _session); + await _pipeline.Received(1).FlattenAndValidateAsync(3, Arg.Any(), 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()); + } + + [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()); + } + + [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(), Arg.Any(), Arg.Any()) + .Returns(Result.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(), Arg.Any(), Arg.Any()); + } + + [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()) + .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(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task GetInstanceStalenessAsync_EmptyInput_ReturnsEmptyWithoutReads() + { + var result = await _service.GetInstanceStalenessAsync([]); + + Assert.Empty(result); + await _repo.DidNotReceive().GetDeployedRevisionHashesAsync(Arg.Any()); + } +}