fix(deployment): pending-updates tick uses a slim instance projection, hash-only flatten, one untracked mismatch snapshot read and a per-tick secret lookup memo (review 2026-10-21 04/P2)

2026-10-22 06:39:45 -04:00
parent 971a9886b8
commit df67fabe68
17 changed files with 443 additions and 128 deletions
@@ -383,6 +383,17 @@ public interface IDeploymentManagerRepository
        IReadOnlyList<PendingDeployment> candidates, DateTimeOffset nowUtc,
        CancellationToken cancellationToken = default);

+    /// <summary>
+    /// Lists every instance that is not <see cref="Types.Enums.InstanceState.NotDeployed"/> as a
+    /// slim, untracked projection — only <c>Id</c>, <c>UniqueName</c>, <c>SiteId</c>,
+    /// <c>TemplateId</c> and <c>State</c> are populated; override and binding collections are
+    /// empty. For the pending-updates sweep, which needs only identity, site and state, and
+    /// used to load every instance with all four override collections, tracked.
+    /// </summary>
+    /// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
+    /// <returns>The deployed (Enabled or Disabled) instances, as slim projections.</returns>
+    Task<IReadOnlyList<Instance>> GetDeployedInstanceSummariesAsync(CancellationToken cancellationToken = default);
+
    // Instance lookups for deployment pipeline
    /// <summary>
    /// Gets an instance by ID.
@@ -724,6 +724,23 @@ public class DeploymentManagerRepository : IDeploymentManagerRepository
                    r.CreatedAtUtc, r.ExpiresAtUtc) { Id = r.Id });
    }

+    /// <inheritdoc />
+    public async Task<IReadOnlyList<Instance>> GetDeployedInstanceSummariesAsync(
+        CancellationToken cancellationToken = default)
+    {
+        return await _dbContext.Set<Instance>()
+            .AsNoTracking()
+            .Where(i => i.State != InstanceState.NotDeployed)
+            .Select(i => new Instance(i.UniqueName)
+            {
+                Id = i.Id,
+                SiteId = i.SiteId,
+                TemplateId = i.TemplateId,
+                State = i.State
+            })
+            .ToListAsync(cancellationToken);
+    }
+
    // --- Instance lookups for deployment pipeline ---

    /// <inheritdoc />
@@ -9,6 +9,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
 using ZB.MOM.WW.ScadaBridge.Commons.Types;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Secrets;
 using ZB.MOM.WW.ScadaBridge.Communication;
+using ZB.MOM.WW.Secrets.Abstractions;

 namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;

@@ -152,7 +153,8 @@ public class ArtifactDeploymentService
        int siteId,
        GlobalArtifactSnapshot globals,
        CancellationToken cancellationToken,
-        string? deploymentId)
+        string? deploymentId,
+        IDictionary<string, StoredSecret?>? secretLookupCache = null)
    {
        var dataConnections = await _siteRepo.GetDataConnectionsBySiteIdAsync(siteId, cancellationToken);

@@ -160,7 +162,8 @@ public class ArtifactDeploymentService
        var dataConnectionArtifacts = dataConnections.Select(dc =>
            new DataConnectionArtifact(dc.Name, dc.Protocol, dc.PrimaryConfiguration, dc.BackupConfiguration, dc.FailoverRetryCount)).ToList();

-        var collected = await _secretsCollector.CollectAsync(dataConnectionArtifacts, cancellationToken);
+        var collected = await _secretsCollector.CollectAsync(
+            dataConnectionArtifacts, cancellationToken, secretLookupCache);

        var command = new DeployArtifactsCommand(
            deploymentId ?? Guid.NewGuid().ToString("N"),
@@ -439,10 +442,12 @@ public class ArtifactDeploymentService
        // a non-empty Missing list means this site's command must never reach
        // _communicationService — see the fan-out loop below.
        var siteCommands = new Dictionary<int, (DeployArtifactsCommand Command, IReadOnlyList<SecretReference> Missing, string? SecretStoreError)>();
+        // One secret-store lookup per referenced name for the whole sweep, not per site.
+        var secretLookups = new Dictionary<string, StoredSecret?>(StringComparer.Ordinal);
        foreach (var site in sites)
        {
            siteCommands[site.Id] = await BuildDeployArtifactsCommandAsync(
-                site.Id, globals, cancellationToken, deploymentId);
+                site.Id, globals, cancellationToken, deploymentId, secretLookups);
        }

        // Deploy to each site in parallel with per-site timeout.
@@ -723,8 +728,9 @@ public class ArtifactDeploymentService
    /// </returns>
    /// <remarks>
    /// The global artifact queries run ONCE for the whole sweep (the same hoist
-    /// <see cref="DeployToAllSitesAsync"/> performs); only the per-site data-connection query
-    /// runs per site. The commands are built with the constant deployment id <c>"current"</c>,
+    /// <see cref="DeployToAllSitesAsync"/> performs), and each referenced secret is read from
+    /// the central store once per call however many sites reference it; only the per-site
+    /// data-connection query runs per site. The commands are built with the constant deployment id <c>"current"</c>,
    /// which is harmless because <see cref="ArtifactRevisionHashService"/> excludes both the
    /// deployment id and the command timestamp from the hash — that exclusion is what makes a
    /// computed-now hash comparable with one stamped by a past deploy.
@@ -751,6 +757,10 @@ public class ArtifactDeploymentService
            return hashes;

        var globals = await FetchGlobalArtifactsAsync(cancellationToken);
+
+        // Per-evaluation memo: a secret referenced by N sites is read from the central store
+        // once per pending-updates tick, not once per site.
+        var secretLookups = new Dictionary<string, StoredSecret?>(StringComparer.Ordinal);
        foreach (var site in sites)
        {
            // The Missing tuple element is deliberately unused here: ComputeHash reads
@@ -758,7 +768,7 @@ public class ArtifactDeploymentService
            // reference still folds a distinguishable marker into the hash — the site reads
            // *Out of date* instead of this call failing the whole batch.
            var (command, _, secretStoreError) = await BuildDeployArtifactsCommandAsync(
-                site.Id, globals, cancellationToken, deploymentId: "current");
+                site.Id, globals, cancellationToken, deploymentId: "current", secretLookups);

            // A central secret-store outage leaves this site's current hash
            // unknowable. Omitting it from the map makes the site read *Out of date* — the
@@ -1811,12 +1811,15 @@ public class DeploymentService
    /// <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
+    /// walked once (the same trick <see cref="DeploySiteAsync"/> uses), and each one is
+    /// hash-only (<see cref="IFlatteningPipeline.ComputeRevisionHashAsync"/>) — no
+    /// validation pass whose result would be discarded. 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
+    /// projected hash DISAGREES with the re-flatten is the snapshot read — in ONE
+    /// untracked batch for all such instances — 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>
    ///
@@ -1862,6 +1865,10 @@ public class DeploymentService
        var firstFailureInstanceId = 0;
        string? firstFailureError = null;

+        // Instances whose re-flattened hash disagrees with the projected deployed hash,
+        // resolved together after the loop with ONE untracked snapshot read.
+        var mismatched = new Dictionary<int, string>();
+
        foreach (var instanceId in instanceIds.Distinct())
        {
            cancellationToken.ThrowIfCancellationRequested();
@@ -1871,10 +1878,10 @@ public class DeploymentService

            evaluated++;

-            // 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);
+            // Read-only path: hash only — no validation (its result was always discarded
+            // here), so no shared-script/schema loads and no structural/semantic pass.
+            var currentResult = await _flatteningPipeline.ComputeRevisionHashAsync(
+                instanceId, session, cancellationToken);
            if (currentResult.IsFailure)
            {
                _logger.LogDebug(
@@ -1891,22 +1898,34 @@ public class DeploymentService
                continue;
            }

-            var currentHash = currentResult.Value.RevisionHash;
+            var currentHash = currentResult.Value;
            if (string.Equals(currentHash, deployedHash, StringComparison.Ordinal))
            {
                staleness[instanceId] = false;
                continue;
            }

+            mismatched[instanceId] = currentHash;
+        }
+
+        if (mismatched.Count > 0)
+        {
            // 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;
+            // One untracked batch read — the snapshots' JSON is hashed and dropped,
+            // never held by the scoped change tracker.
+            var snapshots = await _repository.GetDeployedSnapshotsByInstanceIdsAsync(
+                mismatched.Keys.ToList(), cancellationToken);
+            foreach (var (instanceId, currentHash) in mismatched)
+            {
+                // Vanished between the projection read and now: undeployed concurrently.
+                if (!snapshots.TryGetValue(instanceId, out var snapshot))
+                    continue;

-            var normalizedDeployedHash = ComputeDeployedHash(snapshot, instanceId, out _);
-            staleness[instanceId] =
-                !string.Equals(normalizedDeployedHash, currentHash, StringComparison.Ordinal);
+                var normalizedDeployedHash = ComputeDeployedHash(snapshot, instanceId, out _);
+                staleness[instanceId] =
+                    !string.Equals(normalizedDeployedHash, currentHash, StringComparison.Ordinal);
+            }
        }

        if (failedCount > 0)
@@ -76,30 +76,11 @@ public class FlatteningPipeline : IFlatteningPipeline
        // those to one load each.
        session ??= CreateSession();

-        // Load instance with full graph
-        var instance = await _templateRepo.GetInstanceByIdAsync(instanceId, cancellationToken);
-        if (instance == null)
-            return Result<FlatteningPipelineResult>.Failure($"Instance with ID {instanceId} not found.");
+        var flattened = await FlattenCoreAsync(instanceId, session, cancellationToken);
+        if (flattened.IsFailure)
+            return Result<FlatteningPipelineResult>.Failure(flattened.Error);

-        var graph = await LoadTemplateGraphAsync(instance.TemplateId, session, cancellationToken);
-        if (graph is null)
-            return Result<FlatteningPipelineResult>.Failure("Template chain is empty.");
-
-        var (templateChain, compositionMap, composedChains) = graph.Value;
-
-        // Load data connections for the site
-        var dataConnections = await session.GetDataConnectionsAsync(
-            instance.SiteId,
-            siteId => LoadDataConnectionsAsync(siteId, cancellationToken));
-
-        // Flatten
-        var flattenResult = _flatteningService.Flatten(
-            instance, templateChain, compositionMap, composedChains, dataConnections);
-
-        if (flattenResult.IsFailure)
-            return Result<FlatteningPipelineResult>.Failure(flattenResult.Error);
-
-        var config = flattenResult.Value;
+        var (config, dataConnections) = flattened.Value;

        // Load shared scripts for semantic validation. Session-global: identical
        // for every instance in a batch, so hoisted to one query per session
@@ -180,6 +161,65 @@ public class FlatteningPipeline : IFlatteningPipeline
            new FlatteningPipelineResult(config, hash, validation));
    }

+    /// <inheritdoc />
+    public async Task<Result<string>> ComputeRevisionHashAsync(
+        int instanceId,
+        FlattenSession? session = null,
+        CancellationToken cancellationToken = default)
+    {
+        session ??= CreateSession();
+
+        // Flatten + hash only. The read-only staleness callers discard the validation result
+        // (a flatten whose validation has errors still has a hash, and FlattenAndValidateAsync
+        // returns it), so the structural/semantic validation, and the shared-script and
+        // schema-library loads that exist only to feed it, are pure cost here. The hash is
+        // computed over the same flattened config by the same service, so it is identical to
+        // FlattenAndValidateAsync's RevisionHash.
+        var flattened = await FlattenCoreAsync(instanceId, session, cancellationToken);
+        if (flattened.IsFailure)
+            return Result<string>.Failure(flattened.Error);
+
+        return Result<string>.Success(_revisionHashService.ComputeHash(flattened.Value.Config));
+    }
+
+    /// <summary>
+    /// The part of the pipeline every caller needs: load the instance and its template graph
+    /// (through <paramref name="session"/>), resolve the site's data connections, and flatten.
+    /// </summary>
+    /// <returns>The flattened configuration plus the data connections it was resolved against.</returns>
+    private async Task<Result<(FlattenedConfiguration Config, IReadOnlyDictionary<int, DataConnection> DataConnections)>>
+        FlattenCoreAsync(int instanceId, FlattenSession session, CancellationToken cancellationToken)
+    {
+        // Load instance with full graph
+        var instance = await _templateRepo.GetInstanceByIdAsync(instanceId, cancellationToken);
+        if (instance == null)
+            return Result<(FlattenedConfiguration, IReadOnlyDictionary<int, DataConnection>)>.Failure(
+                $"Instance with ID {instanceId} not found.");
+
+        var graph = await LoadTemplateGraphAsync(instance.TemplateId, session, cancellationToken);
+        if (graph is null)
+            return Result<(FlattenedConfiguration, IReadOnlyDictionary<int, DataConnection>)>.Failure(
+                "Template chain is empty.");
+
+        var (templateChain, compositionMap, composedChains) = graph.Value;
+
+        // Load data connections for the site
+        var dataConnections = await session.GetDataConnectionsAsync(
+            instance.SiteId,
+            siteId => LoadDataConnectionsAsync(siteId, cancellationToken));
+
+        // Flatten
+        var flattenResult = _flatteningService.Flatten(
+            instance, templateChain, compositionMap, composedChains, dataConnections);
+
+        if (flattenResult.IsFailure)
+            return Result<(FlattenedConfiguration, IReadOnlyDictionary<int, DataConnection>)>.Failure(
+                flattenResult.Error);
+
+        return Result<(FlattenedConfiguration, IReadOnlyDictionary<int, DataConnection>)>.Success(
+            (flattenResult.Value, dataConnections));
+    }
+
    /// <inheritdoc />
    public async Task<Result<FlattenedConfiguration>> PreviewAsync(
        int templateId,
@@ -39,6 +39,28 @@ public interface IFlatteningPipeline
        bool validateScripts = true,
        FlattenSession? session = null);

+    /// <summary>
+    /// Flattens an instance and returns ONLY its revision hash — no validation, and none of
+    /// the shared-script / schema-library loads that feed it. For the read-only staleness
+    /// paths (the pending-updates sweep, the bulk-deploy up-to-date pre-filter,
+    /// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport.IStaleInstanceProbe"/>),
+    /// which only ever compare the hash. The hash equals
+    /// <see cref="FlatteningPipelineResult.RevisionHash"/> from
+    /// <see cref="FlattenAndValidateAsync"/> for the same configuration (a flatten that
+    /// fails validation still has a hash there too).
+    /// </summary>
+    /// <param name="instanceId">Id of the instance to flatten.</param>
+    /// <param name="session">
+    /// Optional shared flatten session, as for <see cref="FlattenAndValidateAsync"/>; pass the
+    /// batch's session so a template chain shared by N instances is walked once.
+    /// </param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>The revision hash, or a failure result if the instance could not be flattened.</returns>
+    Task<Result<string>> ComputeRevisionHashAsync(
+        int instanceId,
+        FlattenSession? session = null,
+        CancellationToken cancellationToken = default);
+
    /// <summary>
    /// Flattens <paramref name="templateId"/> as if an instance with no overrides and no
    /// bindings existed at <paramref name="siteId"/>. Attribute list only — no validation,
@@ -33,7 +33,7 @@ namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.PendingUpdates;
 public sealed class PendingUpdatesEvaluator : IPendingUpdatesEvaluator
 {
    private readonly ISiteRepository _siteRepository;
-    private readonly ITemplateEngineRepository _templateRepository;
+    private readonly IDeploymentManagerRepository _deploymentRepository;
    private readonly ArtifactDeploymentService _artifactDeploymentService;
    private readonly DeploymentService _deploymentService;
    private readonly ILogger<PendingUpdatesEvaluator> _logger;
@@ -42,19 +42,22 @@ public sealed class PendingUpdatesEvaluator : IPendingUpdatesEvaluator
    /// Initializes a new evaluator.
    /// </summary>
    /// <param name="siteRepository">Repository the site list is read from.</param>
-    /// <param name="templateRepository">Repository the instance list is read from.</param>
+    /// <param name="deploymentRepository">
+    /// Repository the slim deployed-instance list (id, site, state — no override collections)
+    /// is read from.
+    /// </param>
    /// <param name="artifactDeploymentService">Computes each site's current artifact revision hash.</param>
    /// <param name="deploymentService">Evaluates instance staleness in one batched flatten session.</param>
    /// <param name="logger">Logger instance.</param>
    public PendingUpdatesEvaluator(
        ISiteRepository siteRepository,
-        ITemplateEngineRepository templateRepository,
+        IDeploymentManagerRepository deploymentRepository,
        ArtifactDeploymentService artifactDeploymentService,
        DeploymentService deploymentService,
        ILogger<PendingUpdatesEvaluator> logger)
    {
        _siteRepository = siteRepository;
-        _templateRepository = templateRepository;
+        _deploymentRepository = deploymentRepository;
        _artifactDeploymentService = artifactDeploymentService;
        _deploymentService = deploymentService;
        _logger = logger;
@@ -66,14 +69,16 @@ public sealed class PendingUpdatesEvaluator : IPendingUpdatesEvaluator
        var stopwatch = Stopwatch.StartNew();

        var sites = await _siteRepository.GetAllSitesAsync(cancellationToken);
-        var instances = await _templateRepository.GetAllInstancesAsync(cancellationToken);
+        // Slim, untracked projection: identity, site and state are all this pass reads.
+        var instances = await _deploymentRepository.GetDeployedInstanceSummariesAsync(cancellationToken);

        var currentArtifactHashes =
            await _artifactDeploymentService.ComputeCurrentArtifactHashesAsync(sites, cancellationToken);

        // Only instances the operator has actually deployed can be stale; a
        // NotDeployed instance has nothing to compare against, and flattening it
-        // would be pure cost on every TTL expiry.
+        // would be pure cost on every TTL expiry. The repository already filters them
+        // out in SQL; the in-memory filter keeps that contract explicit here.
        var deployedInstances = instances.Where(i => i.State != InstanceState.NotDeployed).ToList();
        var staleness = await _deploymentService.GetInstanceStalenessAsync(
            deployedInstances.Select(i => i.Id).ToList(), cancellationToken);
@@ -39,6 +39,13 @@ public sealed class SiteSecretsCollector(ISecretStore store)
    /// </summary>
    /// <param name="connections">The site's data-connection artifacts to scan.</param>
    /// <param name="ct">A token to cancel the operation.</param>
+    /// <param name="lookupCache">
+    /// Optional per-sweep memo of store lookups, keyed by secret name (a <see langword="null"/>
+    /// value records "absent"). A multi-site caller passes ONE dictionary for the whole sweep so
+    /// a secret referenced by N sites is read from the store once, not N times. Only successful
+    /// lookups are recorded; a store read failure is never cached. Not thread-safe — the
+    /// callers build their per-site commands serially.
+    /// </param>
    /// <returns>
    /// The resolved rows (mapped field-for-field from <see cref="StoredSecret"/>) and the
    /// references that were absent or tombstoned in the store. A store read failure is
@@ -48,7 +55,8 @@ public sealed class SiteSecretsCollector(ISecretStore store)
    /// secret never touches the store at all.
    /// </returns>
    public async Task<SiteSecretsCollection> CollectAsync(
-        IEnumerable<DataConnectionArtifact> connections, CancellationToken ct)
+        IEnumerable<DataConnectionArtifact> connections, CancellationToken ct,
+        IDictionary<string, StoredSecret?>? lookupCache = null)
    {
        var secrets = new List<SecretArtifact>();
        var missing = new List<SecretReference>();
@@ -58,7 +66,12 @@ public sealed class SiteSecretsCollector(ISecretStore store)
            StoredSecret? row;
            try
            {
-                row = await store.GetAsync(new SecretName(reference.Name), ct);
+                if (lookupCache is null || !lookupCache.TryGetValue(reference.Name, out row))
+                {
+                    row = await store.GetAsync(new SecretName(reference.Name), ct);
+                    if (lookupCache is not null)
+                        lookupCache[reference.Name] = row;
+                }
            }
            catch (Exception ex) when (!ct.IsCancellationRequested)
            {
@@ -19,9 +19,9 @@ public sealed class StaleInstanceProbe : IStaleInstanceProbe
    /// <inheritdoc />
    public async Task<string?> GetCurrentRevisionHashAsync(int instanceId, CancellationToken cancellationToken = default)
    {
-        var session = _flatteningPipeline.CreateSession();
-        var result = await _flatteningPipeline.FlattenAndValidateAsync(instanceId, cancellationToken,
-            validateScripts: false, session).ConfigureAwait(false);
-        return result.IsSuccess ? result.Value.RevisionHash : null;
+        // Hash only: staleness never reads the validation result.
+        var result = await _flatteningPipeline.ComputeRevisionHashAsync(
+            instanceId, _flatteningPipeline.CreateSession(), cancellationToken).ConfigureAwait(false);
+        return result.IsSuccess ? result.Value : null;
    }
 }