fix(secrets): anti-entropy never ran — Self read after await in the replication actor

I dismissed this finding from the code review as a false positive, reasoning that Akka's ActorBase caches Self in a field and that three passing anti-entropy tests traverse the path. Both premises were wrong. Self resolves through Context, which is [ThreadStatic], and throws NotSupportedException once a continuation resumes on a thread-pool thread. The tests passed because a local SQLite store usually completes await SYNCHRONOUSLY, so the continuation stayed on the mailbox thread and the context was still intact. Correctness therefore depended on store latency and thread-pool timing: green here, broken under a slower or contended store, with the only symptom a per-peer warning every announce interval while nodes silently stopped converging. The live-broadcast fast path masked it further — only the anti-entropy repair path was dead. Captures self on the actor thread and passes it in. Adds GenuinelyAsyncSecretStore to force the async path, a regression test that fails on the unfixed code (20s timeout) and passes in 3s after, and ActorContextAfterAwaitTests pinning the underlying Akka behaviour so the wrong assumption cannot be made again. Audited every remaining Self/Sender access in the actor. 170 pass offline / 184 with the live SQL suite / 1 skip / 0 warnings. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
2026-07-18 04:42:16 -04:00
parent dd0a846b64
commit 15ef1f32a8
6 changed files with 239 additions and 11 deletions
@@ -143,9 +143,11 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers
        }

        IActorRef peer = Sender;
+        // Captured here, on the actor thread, because the exchange awaits and cannot read Self after.
+        IActorRef self = Self;
        IReadOnlyList<SecretManifestEntry> remote = [.. message.Entries.Select(e => e.ToEntry())];

-        ReconcileWithPeerAsync(peer, remote).PipeTo(Self, failure: ex => new ExchangeFailed(ex));
+        ReconcileWithPeerAsync(peer, self, remote).PipeTo(self, failure: ex => new ExchangeFailed(ex));
    }

    private void HandlePullRequest(SecretPullRequest message)
@@ -210,8 +212,19 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers

    // Both halves of an anti-entropy exchange, computed from one manifest: push what the peer is
    // behind on, ask for what we are behind on.
+    //
+    // `self` is a PARAMETER, captured by the caller while the actor context is still current. It
+    // must not be read as the `Self` property below: `Self` resolves through Context, which is
+    // [ThreadStatic], and every `await` here can resume on a thread-pool thread where that context
+    // is gone — reading it there throws NotSupportedException and faults the whole exchange.
+    //
+    // This was a live bug. It hid because a local SQLite store usually completes `await`
+    // synchronously, keeping the continuation on the mailbox thread and the context intact, so the
+    // entire cluster suite passed while anti-entropy would have broken under any slower or more
+    // contended store. Anti_entropy_still_works_when_the_store_is_genuinely_asynchronous forces the
+    // async path and fails if this regresses.
    private async Task<ExchangeDone> ReconcileWithPeerAsync(
-        IActorRef peer, IReadOnlyList<SecretManifestEntry> remote)
+        IActorRef peer, IActorRef self, IReadOnlyList<SecretManifestEntry> remote)
    {
        IReadOnlyList<SecretManifestEntry> localManifest =
            await _store.GetManifestAsync(CancellationToken.None).ConfigureAwait(false);
@@ -226,7 +239,7 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers

            if (rows.Count > 0)
            {
-                peer.Tell(new SecretRowsMessage([.. rows.Select(SecretRowDto.FromStoredSecret)]), Self);
+                peer.Tell(new SecretRowsMessage([.. rows.Select(SecretRowDto.FromStoredSecret)]), self);
            }
        }

@@ -235,7 +248,7 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers

        if (pull.Count > 0)
        {
-            peer.Tell(new SecretPullRequest([.. pull.Select(n => n.Value)]), Self);
+            peer.Tell(new SecretPullRequest([.. pull.Select(n => n.Value)]), self);
        }

        return ExchangeDone.Instance;
@@ -0,0 +1,89 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+
+namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
+
+/// <summary>
+/// Pins the Akka behaviour that <see cref="SecretReplicationActor"/> depends on: after a genuine
+/// thread hop, actor context is <b>gone</b>, so <c>Self</c> cannot be read.
+/// </summary>
+/// <remarks>
+/// <para>
+/// This is here because assuming otherwise produced a real bug. <c>Self</c> resolves through
+/// <c>Context</c>, which is <c>[ThreadStatic]</c>; a continuation that resumes on a thread-pool
+/// thread has none, and reading it throws <see cref="NotSupportedException"/>.
+/// </para>
+/// <para>
+/// What made it dangerous is that it does <em>not</em> reliably throw. When the awaited operation
+/// completes synchronously — which a local SQLite store usually does — the continuation stays on the
+/// actor's mailbox thread and the context is still intact, so the illegal read succeeds. Correctness
+/// then depends on store latency and thread-pool timing. The rule the production code follows is
+/// therefore unconditional: capture <c>Self</c> (and <c>Sender</c>) into locals while still on the
+/// actor thread, and never touch actor context after an <c>await</c>.
+/// </para>
+/// </remarks>
+public sealed class ActorContextAfterAwaitTests : TestKit
+{
+    private sealed record Probe;
+
+    private sealed record Outcome(string? SelfError, string? ContextError, string? CapturedSelfPath);
+
+    private sealed class ProbeActor : ReceiveActor
+    {
+        public ProbeActor()
+        {
+            ReceiveAsync<Probe>(async probe =>
+            {
+                IActorRef replyTo = Sender;
+
+                // The correct pattern: capture while the context is still current.
+                IActorRef capturedSelf = Self;
+
+                // Force a real hop off the mailbox thread.
+                await Task.Delay(50).ConfigureAwait(false);
+                await Task.Yield();
+
+                string? selfError = null;
+                string? contextError = null;
+
+                try
+                {
+                    ActorPath path = Self.Path;
+                    GC.KeepAlive(path);
+                }
+                catch (Exception ex)
+                {
+                    selfError = ex.GetType().Name;
+                }
+
+                try
+                {
+                    IActorRef fromContext = Context.Self;
+                    GC.KeepAlive(fromContext);
+                }
+                catch (Exception ex)
+                {
+                    contextError = ex.GetType().Name;
+                }
+
+                replyTo.Tell(new Outcome(selfError, contextError, capturedSelf.Path.ToString()));
+            });
+        }
+    }
+
+    [Fact]
+    public void Self_and_Context_are_both_unreadable_after_a_real_thread_hop()
+    {
+        IActorRef probe = Sys.ActorOf(Props.Create(() => new ProbeActor()), "context-probe");
+        probe.Tell(new Probe());
+
+        Outcome outcome = ExpectMsg<Outcome>(TimeSpan.FromSeconds(10));
+
+        // Both throw — Self is not a cached field, it goes through the ThreadStatic context.
+        Assert.Equal("NotSupportedException", outcome.SelfError);
+        Assert.Equal("NotSupportedException", outcome.ContextError);
+
+        // ...but a reference captured beforehand stays valid, which is what the actor relies on.
+        Assert.EndsWith("/user/context-probe", outcome.CapturedSelfPath, StringComparison.Ordinal);
+    }
+}
@@ -0,0 +1,84 @@
+using ZB.MOM.WW.Secrets.Abstractions;
+
+namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests.Fakes;
+
+/// <summary>
+/// Wraps a store so every operation genuinely yields the thread before completing.
+/// </summary>
+/// <remarks>
+/// <para>
+/// This exists to close a blind spot that let a real bug through. A local SQLite store is fast
+/// enough that <c>await</c> on it frequently completes <b>synchronously</b>, which leaves the
+/// caller on the actor's own mailbox thread — and therefore leaves the actor context (a
+/// <c>[ThreadStatic]</c>) intact. Code that illegally touches actor context after an <c>await</c>
+/// then works by accident, and the whole suite passes.
+/// </para>
+/// <para>
+/// Under a slower store, a contended one, or simply a differently-timed thread pool, the same
+/// <c>await</c> goes properly asynchronous, the continuation lands on a pool thread with no actor
+/// context, and the code throws. That is the worst shape of bug: green in tests, broken in
+/// production, timing-dependent either way. Forcing the async path makes it deterministic.
+/// </para>
+/// </remarks>
+/// <param name="inner">The store to wrap.</param>
+public sealed class GenuinelyAsyncSecretStore(ISecretStore inner) : ISecretStore
+{
+    private static async Task YieldRealAsync()
+    {
+        // Task.Yield alone can still be scheduled inline by some schedulers; a real (if tiny) delay
+        // guarantees the continuation is posted to the thread pool.
+        await Task.Delay(5).ConfigureAwait(false);
+        await Task.Yield();
+    }
+
+    /// <inheritdoc />
+    public async Task<StoredSecret?> GetAsync(SecretName name, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        return await inner.GetAsync(name, ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task UpsertAsync(StoredSecret row, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        await inner.UpsertAsync(row, ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task<bool> DeleteAsync(SecretName name, string? actor, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        return await inner.DeleteAsync(name, actor, ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<SecretMetadata>> ListAsync(bool includeDeleted, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        return await inner.ListAsync(includeDeleted, ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        return await inner.GetManifestAsync(ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task ApplyReplicatedAsync(StoredSecret row, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        await inner.ApplyReplicatedAsync(row, ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async Task<bool> ApplyRewrapAsync(
+        StoredSecret rewrappedRow, byte[] expectedCurrentWrappedDek, CancellationToken ct)
+    {
+        await YieldRealAsync().ConfigureAwait(false);
+        return await inner.ApplyRewrapAsync(rewrappedRow, expectedCurrentWrappedDek, ct)
+            .ConfigureAwait(false);
+    }
+}
@@ -4,6 +4,7 @@ using Akka.Configuration;
 using Microsoft.Extensions.Logging.Abstractions;
 using ZB.MOM.WW.Secrets.Abstractions;
 using ZB.MOM.WW.Secrets.Replication;
+using ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests.Fakes;
 using ZB.MOM.WW.Secrets.Sqlite;

 namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
@@ -263,6 +264,34 @@ public sealed class TwoNodeClusterReplicationTests : IAsyncLifetime
            "Node A never learned about a secret that originated on node B.");
    }

+    [Fact]
+    public async Task Anti_entropy_still_works_when_the_store_is_genuinely_asynchronous()
+    {
+        // Regression guard for a bug the rest of this suite could not see. A local SQLite store
+        // usually completes `await` synchronously, so continuations stayed on the actor's mailbox
+        // thread and code that touched actor context after an await worked by accident. Wrapping the
+        // store so every call really yields makes that illegal access deterministic — this test
+        // fails outright if anti-entropy touches actor context across an await.
+        SqliteSecretStore backing = CreateStore();
+        var slowStore = new GenuinelyAsyncSecretStore(backing);
+
+        _systemB.ActorOf(
+            SecretReplicationActor.Props(slowStore, null, AnnounceInterval), "slow-node");
+
+        await _writableA.UpsertAsync(Row("slow/inbound", 0x42), CancellationToken.None);
+
+        Assert.True(
+            await WaitForAsync(slowStore, new SecretName("slow/inbound"), r => r.Ciphertext[0] == 0x42),
+            "A node whose store is genuinely async never received the secret.");
+
+        // And the other direction, which can ONLY converge through the anti-entropy exchange.
+        await backing.UpsertAsync(Row("slow/outbound", 0x43), CancellationToken.None);
+
+        Assert.True(
+            await WaitForAsync(_storeA, new SecretName("slow/outbound"), r => r.Ciphertext[0] == 0x43),
+            "Node A never learned about a secret held by a node whose store is genuinely async.");
+    }
+
    [Fact]
    public async Task A_node_that_joins_late_catches_up_on_everything()
    {