fix(site-secrets): mirror heals same-revision local drift and isolates unreadable rows

Final review I1: a name already ledgered at the shipped revision is now read back every pass; a missing, tombstoned or edited local row is healed through the mirror-owned re-stamp path (bump, apply, read back, invalidate) before verify, instead of a local delete being misreported as a KEK mismatch that blocked every later deploy. Drift logs Warning once per occurrence. M4: row_json that cannot be parsed becomes a Failed entry for that name via the new SiteStorageService.GetDeployedSecretsTolerantAsync; an unreadable removed row is still honoured by name. M10: invalidator ctor doc corrected.
2026-09-28 21:08:34 -04:00
parent 9835ef8be3
commit 6f1787abe1
3 changed files with 277 additions and 27 deletions
@@ -780,9 +780,32 @@ public class SiteStorageService
    /// <summary>
    /// Returns every row in <c>deployed_secrets</c>, including soft-deleted (<c>is_removed</c>)
    /// ones, with each row's <see cref="SecretArtifact"/> round-tripped from <c>row_json</c>.
+    /// Strict: a row whose <c>row_json</c> cannot be parsed throws. The per-node mirror reads
+    /// through <see cref="GetDeployedSecretsTolerantAsync"/> instead, so one unreadable row is
+    /// reported against its own name rather than aborting the whole reconcile pass.
    /// </summary>
    /// <returns>A task that resolves to every stored deployed-secret row.</returns>
    public async Task<IReadOnlyList<DeployedSecretRow>> GetDeployedSecretsAsync()
+    {
+        var (rows, unreadable) = await GetDeployedSecretsTolerantAsync();
+        if (unreadable.Count > 0)
+        {
+            throw new InvalidOperationException(
+                $"deployed_secrets row '{unreadable[0].Name}' has invalid row_json ({unreadable[0].Error})");
+        }
+
+        return rows;
+    }
+
+    /// <summary>
+    /// Returns every row in <c>deployed_secrets</c> like <see cref="GetDeployedSecretsAsync"/>,
+    /// but never throws for a single bad row: a row whose <c>row_json</c> does not deserialize
+    /// is returned in <c>Unreadable</c> (name, removal flag and the parse error — never the
+    /// row's content) instead of failing the whole read (final review M4, per-row isolation).
+    /// </summary>
+    /// <returns>The parsed rows and the rows that could not be parsed.</returns>
+    public async Task<(IReadOnlyList<DeployedSecretRow> Rows, IReadOnlyList<UnreadableDeployedSecretRow> Unreadable)>
+        GetDeployedSecretsTolerantAsync()
    {
        await using var connection = OpenConnection();

@@ -790,22 +813,39 @@ public class SiteStorageService
        command.CommandText = "SELECT name, row_json, is_removed, deployment_id FROM deployed_secrets";

        var results = new List<DeployedSecretRow>();
+        var unreadable = new List<UnreadableDeployedSecretRow>();
        await using var reader = await command.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            var name = reader.GetString(0);
-            var rowJson = reader.GetString(1);
-            var secret = JsonSerializer.Deserialize<SecretArtifact>(rowJson)
-                ?? throw new InvalidOperationException($"deployed_secrets row '{name}' has invalid row_json");
+            var isRemoved = reader.GetInt64(2) != 0;

-            results.Add(new DeployedSecretRow(
-                name,
-                secret,
-                reader.GetInt64(2) != 0,
-                reader.GetString(3)));
+            SecretArtifact? secret;
+            string? error = null;
+            try
+            {
+                secret = JsonSerializer.Deserialize<SecretArtifact>(reader.GetString(1));
+                if (secret is null)
+                {
+                    error = "row_json deserialized to null";
+                }
+            }
+            catch (Exception ex) when (ex is JsonException or NotSupportedException or InvalidCastException)
+            {
+                secret = null;
+                error = $"{ex.GetType().Name}: {ex.Message}";
+            }
+
+            if (secret is null)
+            {
+                unreadable.Add(new UnreadableDeployedSecretRow(name, isRemoved, error!));
+                continue;
+            }
+
+            results.Add(new DeployedSecretRow(name, secret, isRemoved, reader.GetString(3)));
        }

-        return results;
+        return (results, unreadable);
    }

    // ── Mirrored Secrets ledger (node-local only — deliberately NOT LocalDb-replicated) ──
@@ -1331,6 +1371,14 @@ public record NativeAlarmRow(
 /// </summary>
 public record DeployedSecretRow(string Name, SecretArtifact Secret, bool IsRemoved, string DeploymentId);

+/// <summary>
+/// A <c>deployed_secrets</c> row whose <c>row_json</c> could not be parsed — returned by
+/// <see cref="SiteStorageService.GetDeployedSecretsTolerantAsync"/> so the mirror can report it
+/// against its own name instead of aborting the pass. <paramref name="Error"/> is the parse error
+/// (exception type and message), never the row's content.
+/// </summary>
+public record UnreadableDeployedSecretRow(string Name, bool IsRemoved, string Error);
+
 /// <summary>
 /// A single row from the site-local, node-local (NOT LocalDb-replicated) <c>mirrored_secrets</c>
 /// ledger — the names this node has mirrored into its own secret store, and at what revision.
@@ -94,6 +94,12 @@ public sealed record MirrorResult(
 /// that the deployed revision never took effect).
 /// </item>
 /// </list>
+/// <b>Drift at the same revision (final review I1).</b> A name already ledgered at the shipped
+/// revision is still read back every pass: if an operator deleted it locally, or edited its value,
+/// the local row no longer matches the shipped ciphertext and is healed exactly like the owned
+/// re-stamp case above (bumped stamp, apply, read-back, cache invalidation) before the verify —
+/// so a local delete is repaired rather than misreported as a master-key mismatch, and a local
+/// edit is reverted to the deployed value.
 /// Ledgering always uses the SHIPPED artifact's own revision (not any bumped local revision) so
 /// a later pass's "is this ledgered at the shipped revision" check stays correct.
 /// </para>
@@ -111,8 +117,8 @@ public sealed record MirrorResult(
 /// <see cref="ISecretCacheInvalidator"/>; every ZB.MOM.WW.Secrets write path invalidates after a
 /// write for exactly this reason. This mirror does the same: after every successful apply
 /// (including a tombstone re-materialization) or delete, and before the verify step, it calls
-/// <see cref="ISecretCacheInvalidator.Invalidate"/> (injected optionally — <c>null</c> if the
-/// host never registered one) so neither the verify nor a script running immediately afterward
+/// <see cref="ISecretCacheInvalidator.Invalidate"/> (a <c>null</c> invalidator is tolerated for
+/// direct construction; under DI <c>AddZbSecrets</c> always registers one) so neither the verify nor a script running immediately afterward
 /// can observe a stale pre-rotation plaintext for up to the cache TTL.
 /// </para>
 /// <para>
@@ -172,14 +178,23 @@ public sealed class DeployedSecretsMirror : BackgroundService
    /// </summary>
    private readonly Dictionary<string, string> _lastLoggedFailure = new(StringComparer.Ordinal);

+    /// <summary>
+    /// Names whose same-revision local drift (final review I1) has already been logged at
+    /// Warning and not yet healed — a heal that keeps failing logs its drift at Debug on repeats,
+    /// matching the <see cref="_lastLoggedFailure"/> de-spam rule. Cleared on a successful apply.
+    /// </summary>
+    private readonly HashSet<string> _driftLogged = new(StringComparer.Ordinal);
+
    /// <summary>Initializes a new instance of <see cref="DeployedSecretsMirror"/>.</summary>
    /// <param name="storage">Site-local storage exposing <c>deployed_secrets</c> and the <c>mirrored_secrets</c> ledger.</param>
    /// <param name="store">This node's own secret store.</param>
    /// <param name="resolver">This node's own secret resolver, used to trial-decrypt each deployed secret.</param>
    /// <param name="invalidator">
-    /// Optional cache invalidator for <paramref name="resolver"/> (the concrete
-    /// <c>DefaultSecretResolver</c> implements this itself; <c>null</c> if the host never
-    /// registered one, in which case no invalidation is attempted).
+    /// Cache invalidator for <paramref name="resolver"/> (the concrete
+    /// <c>DefaultSecretResolver</c> implements this itself). The parameter has no default, so
+    /// under MS DI activation fails unless an <see cref="ISecretCacheInvalidator"/> is registered
+    /// — <c>AddZbSecrets</c> always registers one. A caller constructing the mirror directly may
+    /// pass <see langword="null"/>, in which case no invalidation is attempted.
    /// </param>
    /// <param name="options">Supplies <see cref="SiteRuntimeOptions.DeployedSecretsMirrorInterval"/>.</param>
    /// <param name="logger">Logger instance. Never logs secret values — names, revisions, and KEK ids only.</param>
@@ -217,7 +232,10 @@ public sealed class DeployedSecretsMirror : BackgroundService
        await _gate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
-            var rows = await _storage.GetDeployedSecretsAsync().ConfigureAwait(false);
+            // Tolerant read (final review M4): a row whose row_json cannot be parsed is handed
+            // back separately instead of throwing out of the whole pass, so it is reported
+            // against its own name like any other per-row failure.
+            var (rows, unreadable) = await _storage.GetDeployedSecretsTolerantAsync().ConfigureAwait(false);
            var mirrored = (await _storage.GetMirroredSecretsAsync().ConfigureAwait(false))
                .ToDictionary(m => m.Name, m => m.Revision, StringComparer.Ordinal);

@@ -232,7 +250,7 @@ public sealed class DeployedSecretsMirror : BackgroundService
                {
                    if (row.IsRemoved)
                    {
-                        await ProcessRemovedRowAsync(row, mirrored, removed, ct).ConfigureAwait(false);
+                        await ProcessRemovedRowAsync(row.Name, mirrored, removed, ct).ConfigureAwait(false);
                    }
                    else
                    {
@@ -250,6 +268,37 @@ public sealed class DeployedSecretsMirror : BackgroundService
                }
            }

+            foreach (var bad in unreadable)
+            {
+                try
+                {
+                    if (bad.IsRemoved)
+                    {
+                        // A removal needs only the name, so an unreadable tombstone is still
+                        // honoured for a name this node previously mirrored.
+                        await ProcessRemovedRowAsync(bad.Name, mirrored, removed, ct).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        ReportFailure(
+                            bad.Name,
+                            $"unreadable:{bad.Error}",
+                            $"Secret '{bad.Name}' could not be read from deployed_secrets on this node ({bad.Error})",
+                            failed,
+                            failedNames);
+                    }
+                }
+                catch (Exception ex) when (!ct.IsCancellationRequested)
+                {
+                    ReportFailure(
+                        bad.Name,
+                        $"error:unreadable:{ex.GetType().Name}:{ex.Message}",
+                        $"Secret '{bad.Name}' failed to process on this node ({ex.GetType().Name}: {ex.Message})",
+                        failed,
+                        failedNames);
+                }
+            }
+
            var result = new MirrorResult(applied, removed, failed, failedNames);

            if (result.Changed)
@@ -262,7 +311,7 @@ public sealed class DeployedSecretsMirror : BackgroundService
            {
                _logger.LogDebug(
                    "Deployed secrets mirror in sync ({Count} secret(s), {FailedCount} failing)",
-                    rows.Count(r => !r.IsRemoved), failed.Count);
+                    rows.Count(r => !r.IsRemoved) + unreadable.Count(u => !u.IsRemoved), failed.Count);
            }

            return result;
@@ -279,26 +328,28 @@ public sealed class DeployedSecretsMirror : BackgroundService
    /// local secret this mirror never applied is left alone.
    /// </summary>
    private async Task ProcessRemovedRowAsync(
-        DeployedSecretRow row, Dictionary<string, long> mirrored, List<string> removed, CancellationToken ct)
+        string rowName, Dictionary<string, long> mirrored, List<string> removed, CancellationToken ct)
    {
-        if (!mirrored.ContainsKey(row.Name))
+        if (!mirrored.ContainsKey(rowName))
        {
            return;
        }

-        await _store.DeleteAsync(new SecretName(row.Name), DeleteActor, ct).ConfigureAwait(false);
-        _invalidator?.Invalidate(new SecretName(row.Name));
-        await _storage.RemoveMirroredSecretAsync(row.Name).ConfigureAwait(false);
-        removed.Add(row.Name);
-        _lastLoggedFailure.Remove(row.Name);
+        await _store.DeleteAsync(new SecretName(rowName), DeleteActor, ct).ConfigureAwait(false);
+        _invalidator?.Invalidate(new SecretName(rowName));
+        await _storage.RemoveMirroredSecretAsync(rowName).ConfigureAwait(false);
+        removed.Add(rowName);
+        _lastLoggedFailure.Remove(rowName);
        _logger.LogInformation(
            "Removed deployed secret {SecretName} (no longer referenced by this site)",
-            row.Name);
+            rowName);
    }

    /// <summary>
    /// Handles a live <c>deployed_secrets</c> row: applies it (confirmed by read-back) when not
-    /// already ledgered at the shipped revision, then verifies it decrypts — unless the apply
+    /// already ledgered at the shipped revision, or when it is ledgered but the local row has
+    /// since gone missing, been tombstoned or drifted to different ciphertext (healed via the
+    /// mirror-owned re-stamp path), then verifies it decrypts — unless the apply
    /// outcome makes a verify meaningless (an out-of-band local secret won LWW), in which case a
    /// single, distinct failure is reported instead.
    /// </summary>
@@ -319,7 +370,29 @@ public sealed class DeployedSecretsMirror : BackgroundService
        var ownedByMirror = mirrored.TryGetValue(row.Name, out var mirroredRevision);
        var alreadyLedgered = ownedByMirror && mirroredRevision == row.Secret.Revision;

-        if (!alreadyLedgered)
+        // Ledgered at the shipped revision is NOT proof the local row still holds it (final
+        // review I1): an operator may have deleted or edited this mirror-owned name on the node
+        // since. Read it back; only a live row whose ciphertext matches the shipped artifact goes
+        // straight to verify. A missing, tombstoned or drifted row is healed through the same
+        // mirror-owned re-stamp path as any other owned mismatch — central is authoritative for
+        // every name it deploys — instead of being misreported as a master-key mismatch.
+        var needsApply = !alreadyLedgered;
+        if (alreadyLedgered)
+        {
+            var local = await _store.GetAsync(name, ct).ConfigureAwait(false);
+            if (local is null || local.IsDeleted || !MatchesShippedCiphertext(local, ToStored(row.Secret)))
+            {
+                needsApply = true;
+                _logger.Log(
+                    _driftLogged.Add(row.Name) ? LogLevel.Warning : LogLevel.Debug,
+                    "Deployed secret {SecretName} r{Revision} was {Drift} on this node; re-applying the deployed revision",
+                    row.Name,
+                    row.Secret.Revision,
+                    local is null ? "missing" : local.IsDeleted ? "deleted" : "changed");
+            }
+        }
+
+        if (needsApply)
        {
            var outcome = await ApplyAndConfirmAsync(row, name, ownedByMirror, ct).ConfigureAwait(false);

@@ -330,6 +403,7 @@ public sealed class DeployedSecretsMirror : BackgroundService
                    _invalidator?.Invalidate(name);
                    applied.Add(row.Name);
                    _lastLoggedFailure.Remove(row.Name);
+                    _driftLogged.Remove(row.Name);
                    _logger.LogInformation(
                        "Applied deployed secret {SecretName} r{Revision} (kek {KekId})",
                        row.Name, row.Secret.Revision, row.Secret.KekId);
@@ -205,6 +205,126 @@ public class DeployedSecretsMirrorTests : IDisposable
        Assert.Equal(2, store.Rows["a"].Revision);
    }

+    // ── Same-revision drift (final review I1) ──
+
+    [Fact]
+    public async Task Reconcile_OwnedNameDeletedLocally_IsHealedAndVerified_NotMisreportedAsKekMismatch()
+    {
+        var time = new FakeTimeProvider(new DateTimeOffset(2026, 9, 28, 18, 0, 0, TimeSpan.Zero));
+        var (mirror, storage, store, resolver, invalidator) = await CreateAsync(time: time);
+        resolver.BackingStore = store; // resolve returns null once the local row is gone
+        await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
+        Assert.Equal(["a"], (await mirror.ReconcileAsync(CancellationToken.None)).Applied);
+
+        // An operator deletes the mirror-owned name on this node (library CLI/UI) — the
+        // ledger still says r1 is mirrored, and central still ships r1.
+        await store.DeleteAsync(new SecretName("a"), "operator", CancellationToken.None);
+        Assert.True(store.IsTombstoned("a"));
+        invalidator.Invalidated.Clear();
+
+        var r = await mirror.ReconcileAsync(CancellationToken.None);
+
+        Assert.Empty(r.Failed);
+        Assert.Equal(["a"], r.Applied);
+        Assert.True(store.Contains("a"));
+        Assert.Equal(new byte[] { 1, 2, 3, 4, 5 }, store.Rows["a"].Ciphertext);
+        Assert.Contains("a", invalidator.Invalidated);
+
+        // Healed state is stable: the next pass is a no-op that still verifies clean.
+        var next = await mirror.ReconcileAsync(CancellationToken.None);
+        Assert.False(next.Changed);
+        Assert.Empty(next.Failed);
+        Assert.Equal(1, (await storage.GetMirroredSecretsAsync()).Single(m => m.Name == "a").Revision);
+    }
+
+    [Fact]
+    public async Task Reconcile_OwnedNameEditedLocally_IsRevertedToTheDeployedCiphertext()
+    {
+        var time = new FakeTimeProvider(new DateTimeOffset(2026, 9, 28, 18, 0, 0, TimeSpan.Zero));
+        var (mirror, storage, store, _, invalidator) = await CreateAsync(time: time);
+        await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
+        Assert.Equal(["a"], (await mirror.ReconcileAsync(CancellationToken.None)).Applied);
+
+        // An operator overwrites the mirror-owned name locally with a NEWER stamp.
+        store.Seed("a",
+            updatedUtc: new DateTimeOffset(2026, 9, 28, 12, 0, 0, TimeSpan.Zero),
+            revision: 7,
+            ciphertext: [200, 201, 202]);
+        invalidator.Invalidated.Clear();
+
+        var r = await mirror.ReconcileAsync(CancellationToken.None);
+
+        Assert.Empty(r.Failed);
+        Assert.Equal(["a"], r.Applied);
+        Assert.Equal(new byte[] { 1, 2, 3, 4, 5 }, store.Rows["a"].Ciphertext);
+        Assert.Equal(new byte[] { 30, 31, 32, 33, 34, 35 }, store.Rows["a"].WrappedDek);
+        Assert.True(store.Rows["a"].Revision > 7); // re-stamped past the operator's row
+        Assert.Contains("a", invalidator.Invalidated);
+    }
+
+    [Fact]
+    public async Task Reconcile_OwnedNameDrift_LogsOncePerDriftNotEveryPass()
+    {
+        var logger = new CapturingLogger<DeployedSecretsMirror>();
+        var time = new FakeTimeProvider(new DateTimeOffset(2026, 9, 28, 18, 0, 0, TimeSpan.Zero));
+        var (mirror, storage, store, _, _) = await CreateAsync(logger: logger, time: time);
+        await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
+        await mirror.ReconcileAsync(CancellationToken.None);
+        await store.DeleteAsync(new SecretName("a"), "operator", CancellationToken.None);
+
+        await mirror.ReconcileAsync(CancellationToken.None);
+        await mirror.ReconcileAsync(CancellationToken.None);
+        await mirror.ReconcileAsync(CancellationToken.None);
+
+        Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning && e.Message.Contains("re-applying"));
+        Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Error);
+    }
+
+    // ── Unreadable row_json (final review M4) ──
+
+    [Fact]
+    public async Task Reconcile_UnparseableRowJson_IsAFailedEntryForThatName_OthersStillApply()
+    {
+        var (mirror, storage, store, _, _) = await CreateAsync();
+        await storage.UpsertDeployedSecretsAsync([Art("bad", 1), Art("good", 1)], "d1");
+        await CorruptRowJsonAsync(storage, "bad");
+
+        var r = await mirror.ReconcileAsync(CancellationToken.None);
+
+        Assert.Equal(["good"], r.Applied);
+        Assert.Equal(["bad"], r.FailedNames);
+        var message = Assert.Single(r.Failed);
+        Assert.Contains("'bad'", message);
+        Assert.Contains("could not be read", message);
+        Assert.True(store.Contains("good"));
+        Assert.False(store.Contains("bad"));
+    }
+
+    [Fact]
+    public async Task Reconcile_UnparseableRemovedRow_StillRemovesAMirroredName()
+    {
+        var (mirror, storage, store, _, _) = await CreateAsync();
+        await storage.UpsertDeployedSecretsAsync([Art("a", 1)], "d1");
+        await mirror.ReconcileAsync(CancellationToken.None);
+        await storage.UpsertDeployedSecretsAsync([], "d2"); // soft-removes "a"
+        await CorruptRowJsonAsync(storage, "a");
+
+        var r = await mirror.ReconcileAsync(CancellationToken.None);
+
+        Assert.Equal(["a"], r.Removed);
+        Assert.Empty(r.Failed);
+        Assert.True(store.IsTombstoned("a"));
+    }
+
+    private static async Task CorruptRowJsonAsync(SiteStorageService storage, string name)
+    {
+        await using var connection = storage.CreateConnection();
+        await using var command = connection.CreateCommand();
+        command.CommandText = "UPDATE deployed_secrets SET row_json = '{not json' WHERE name = @name";
+        command.Parameters.AddWithValue("@name", name);
+        Assert.Equal(1, await command.ExecuteNonQueryAsync());
+    }
+
    [Fact]
    public async Task Reconcile_AppliesThroughApplyReplicatedAsync_MappedFromSecretArtifact()
    {
@@ -740,6 +860,9 @@ public class DeployedSecretsMirrorTests : IDisposable

        public int CallCount;

+        /// <summary>When set, a name that is absent or tombstoned in this store resolves to null, like the real resolver.</summary>
+        public FakeSecretStore? BackingStore { get; set; }
+
        /// <summary>Row absent/tombstoned path — <c>DefaultSecretResolver.GetAsync</c> returns null.</summary>
        public void ReturnNullFor(string name) => _nullReturning.Add(name);

@@ -760,6 +883,11 @@ public class DeployedSecretsMirrorTests : IDisposable
                throw new InvalidOperationException("simulated decrypt failure");
            }

+            if (BackingStore is not null && !BackingStore.Contains(name.Value))
+            {
+                return Task.FromResult<string?>(null);
+            }
+
            return Task.FromResult(_nullReturning.Contains(name.Value) ? null : "decrypted-value");
        }
    }