fix(secrets-akka): announce from the mailbox with a sender; chunk anti-entropy payloads (scadaproj#45)

Secrets-011: HandleAnnounceTick pipes the manifest read back to the actor and publishes the announce from the mailbox with Self as sender. Under a genuinely async store it used to publish from a pool continuation with NoSender, so every peer replied to deadLetters; a faulted manifest read was dropped silently and is now logged. Secrets-012: the exchange's push rows, the pull reply and the pull request are split into messages of at most 64 KB (measured with the wire encoding), half Akka.Remote's default 128000-byte frame, so a cold node converges against a large store. A single over-budget row is still sent alone; the manifest announce is not split (documented in clustered-secrets.md). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 15:16:57 -05:00
parent 5e2d3dd1bc
commit 849640f467
4 changed files with 327 additions and 22 deletions
@@ -171,6 +171,20 @@ One manifest per node per interval: names, revisions, timestamps, tombstone flag
 Cost grows as O(nodes × secrets), so a large cluster with a large secret set should lengthen
 `AnnounceInterval` rather than shorten it.

+### Message size and Akka.Remote's frame limit
+
+Akka.Remote drops any message larger than `akka.remote.dot-netty.tcp.maximum-frame-size` (default
+128000 bytes). Since `0.6.6` the rows an exchange pushes, the rows a pull is answered with, and the
+pull request itself are split into messages of at most ~64 KB each, so a cold node catching up on a
+peer's whole inventory converges however many secrets there are. Two bounds remain:
+
+- **One very large secret** (a single row whose ciphertext alone exceeds the budget, e.g. a large
+  certificate bundle) is still sent as one message. If it is over the frame size it never replicates;
+  raise `maximum-frame-size` on every node.
+- **The manifest announce is not split** — a peer computes what to push against the *whole* manifest.
+  An entry is ~100 bytes, so the default frame holds roughly a thousand secrets. Beyond that, raise
+  `maximum-frame-size`.
+
 ---

 ## Choosing between them
@@ -0,0 +1,65 @@
+using System.Text.Json;
+
+namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Protocol;
+
+/// <summary>
+/// Splits anti-entropy payloads into messages that stay well inside Akka.Remote's frame size
+/// (Secrets-012). A cold node's push or pull set is the peer's whole inventory; sent as one
+/// <see cref="SecretRowsMessage"/> or <see cref="SecretPullRequest"/> it exceeds the default
+/// <c>akka.remote.dot-netty.tcp.maximum-frame-size</c> (128000 bytes) at a few hundred secrets, is
+/// dropped by the transport, and is rebuilt identically every announce — so the node never converges.
+/// </summary>
+/// <remarks>
+/// Each message is sized by the serializer's own encoding (compact STJ JSON) to at most
+/// <see cref="MaxPayloadBytes"/>, half the default frame, leaving room for the remoting envelope.
+/// A single row larger than the budget (a very large secret) is still sent, alone — it cannot be
+/// split, and raising the frame size is then the operator's lever. The manifest announce itself is
+/// not chunked: a receiver computes its push set against the WHOLE remote manifest, so a partial one
+/// would be wrong rather than merely smaller; an entry is a name plus timestamps (~100 bytes), so the
+/// announce fits the default frame up to roughly a thousand secrets.
+/// </remarks>
+internal static class SecretReplicationChunks
+{
+    /// <summary>Byte budget for one serialized message: half the default 128000-byte frame.</summary>
+    internal const int MaxPayloadBytes = 64_000;
+
+    // {"Rows":[ ... ]} / {"Names":[ ... ]} plus slack.
+    private const int EnvelopeBytes = 32;
+
+    /// <summary>Groups <paramref name="rows"/> into messages of at most <paramref name="budgetBytes"/>.</summary>
+    internal static IEnumerable<SecretRowsMessage> Rows(
+        IEnumerable<SecretRowDto> rows, int budgetBytes = MaxPayloadBytes) =>
+        Chunk(rows, row => JsonSerializer.SerializeToUtf8Bytes(row).Length, budgetBytes)
+            .Select(chunk => new SecretRowsMessage(chunk));
+
+    /// <summary>Groups <paramref name="names"/> into pull requests of at most <paramref name="budgetBytes"/>.</summary>
+    internal static IEnumerable<SecretPullRequest> PullRequests(
+        IEnumerable<string> names, int budgetBytes = MaxPayloadBytes) =>
+        Chunk(names, name => JsonSerializer.SerializeToUtf8Bytes(name).Length, budgetBytes)
+            .Select(chunk => new SecretPullRequest(chunk));
+
+    private static IEnumerable<IReadOnlyList<T>> Chunk<T>(IEnumerable<T> items, Func<T, int> size, int budgetBytes)
+    {
+        var current = new List<T>();
+        int used = EnvelopeBytes;
+
+        foreach (T item in items)
+        {
+            int itemBytes = size(item) + 1; // separating comma
+            if (current.Count > 0 && used + itemBytes > budgetBytes)
+            {
+                yield return current;
+                current = new List<T>();
+                used = EnvelopeBytes;
+            }
+
+            current.Add(item);
+            used += itemBytes;
+        }
+
+        if (current.Count > 0)
+        {
+            yield return current;
+        }
+    }
+}
@@ -82,6 +82,9 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers
        Receive<ExchangeFailed>(f =>
            _log.Warning(f.Cause, "Secret anti-entropy exchange with a peer failed; retrying next announce."));
        Receive<ExchangeDone>(_ => { /* successful exchange, piped back to self — nothing to do. */ });
+        Receive<ManifestReady>(HandleManifestReady);
+        Receive<ManifestFailed>(f =>
+            _log.Warning(f.Cause, "Could not read the local secret manifest to announce; retrying next interval."));
    }

    /// <summary>Props for the replication actor.</summary>
@@ -181,9 +184,11 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers

                    if (task.IsCompletedSuccessfully && task.Result.Count > 0)
                    {
-                        peer.Tell(
-                            new SecretRowsMessage([.. task.Result.Select(SecretRowDto.FromStoredSecret)]),
-                            self);
+                        foreach (SecretRowsMessage chunk in
+                                 SecretReplicationChunks.Rows(task.Result.Select(SecretRowDto.FromStoredSecret)))
+                        {
+                            peer.Tell(chunk, self);
+                        }
                    }
                },
                CancellationToken.None,
@@ -191,25 +196,27 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers
                TaskScheduler.Default);
    }

+    // The manifest read is piped back to the mailbox and published from there (Secrets-011). It used
+    // to be published from the read's continuation: under a store whose read really yields, that ran
+    // on a pool thread with no actor context, so the announce carried NoSender, every peer replied to
+    // deadLetters, and anti-entropy between two such nodes never converged. A faulted read used to be
+    // dropped unobserved; now it is logged.
    private void HandleAnnounceTick()
    {
-        _store.GetManifestAsync(CancellationToken.None)
-            .ContinueWith(
-                task =>
-                {
-                    if (task.IsCompletedSuccessfully)
-                    {
-                        _mediator.Tell(new Publish(
-                            Topic,
-                            new SecretManifestAnnounce(
-                                [.. task.Result.Select(SecretManifestEntryDto.FromEntry)])));
-                    }
-                },
-                CancellationToken.None,
-                TaskContinuationOptions.ExecuteSynchronously,
-                TaskScheduler.Default);
+        IActorRef self = Self;
+        _store.GetManifestAsync(CancellationToken.None).PipeTo(
+            self,
+            success: manifest => new ManifestReady(manifest),
+            failure: ex => new ManifestFailed(ex));
    }

+    private void HandleManifestReady(ManifestReady ready) =>
+        _mediator.Tell(
+            new Publish(
+                Topic,
+                new SecretManifestAnnounce([.. ready.Manifest.Select(SecretManifestEntryDto.FromEntry)])),
+            Self);
+
    // 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.
    //
@@ -237,18 +244,18 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers
            IReadOnlyList<StoredSecret> rows =
                await _reconciler.ReadLocalAsync(push, CancellationToken.None).ConfigureAwait(false);

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

        IReadOnlyList<SecretName> pull =
            SecretReplicationReconciler.ComputePullSet(localManifest, remote);

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

        return ExchangeDone.Instance;
@@ -309,6 +316,10 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers

    private sealed record ExchangeFailed(Exception Cause);

+    private sealed record ManifestReady(IReadOnlyList<SecretManifestEntry> Manifest);
+
+    private sealed record ManifestFailed(Exception Cause);
+
    private sealed record ExchangeDone
    {
        public static readonly ExchangeDone Instance = new();
@@ -0,0 +1,215 @@
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.Configuration;
+using Akka.TestKit.Xunit2;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Protocol;
+using ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests.Fakes;
+using ZB.MOM.WW.Secrets.Sqlite;
+
+namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
+
+/// <summary>
+/// Secrets-011 (the periodic announce must carry the actor as sender even under a genuinely async
+/// store, and a failed manifest read must be logged) and Secrets-012 (anti-entropy payloads must be
+/// chunked under Akka.Remote's frame size), driven against one self-joined cluster node.
+/// </summary>
+public sealed class AnnounceAndChunkingTests : TestKit, IDisposable
+{
+    private static readonly TimeSpan AnnounceInterval = TimeSpan.FromMilliseconds(200);
+    private readonly List<string> _dbPaths = [];
+
+    public AnnounceAndChunkingTests()
+        : base(ConfigurationFactory.ParseString("""
+            akka {
+              loglevel = WARNING
+              actor.provider = cluster
+              remote.dot-netty.tcp { hostname = "127.0.0.1", public-hostname = "127.0.0.1", port = 0 }
+            }
+            """).WithFallback(AkkaSecretsReplication.SerializationConfig))
+    {
+        Cluster cluster = Cluster.Get(Sys);
+        cluster.Join(cluster.SelfAddress);
+        AwaitCondition(() => cluster.State.Members.Any(m => m.Status == MemberStatus.Up), TimeSpan.FromSeconds(20));
+    }
+
+    void IDisposable.Dispose()
+    {
+        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+        foreach (string path in _dbPaths)
+        {
+            try { File.Delete(path); } catch (IOException) { }
+        }
+    }
+
+    private SqliteSecretStore CreateStore()
+    {
+        string path = Path.Combine(Path.GetTempPath(), $"zb-akka-chunk-{Guid.NewGuid():N}.db");
+        _dbPaths.Add(path);
+        var factory = new SecretsSqliteConnectionFactory(path);
+        new SqliteSecretsStoreMigrator(factory).MigrateAsync(CancellationToken.None).GetAwaiter().GetResult();
+        return new SqliteSecretStore(factory);
+    }
+
+    private static StoredSecret Row(string name, int ciphertextBytes) => new()
+    {
+        Name = new SecretName(name),
+        ContentType = SecretContentType.Text,
+        Ciphertext = new byte[ciphertextBytes],
+        Nonce = new byte[12],
+        Tag = new byte[16],
+        WrappedDek = new byte[32],
+        WrapNonce = new byte[12],
+        WrapTag = new byte[16],
+        KekId = "sha256:shared-kek",
+        Revision = 0,
+        CreatedUtc = DateTimeOffset.UtcNow,
+        UpdatedUtc = DateTimeOffset.UtcNow,
+    };
+
+    private int SerializedSize(object message) =>
+        new SecretReplicationSerializer((ExtendedActorSystem)Sys).ToBinary(message).Length;
+
+    private void SubscribeTestActor()
+    {
+        DistributedPubSub.Get(Sys).Mediator.Tell(new Subscribe(SecretReplicationActor.Topic, TestActor));
+        ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(10));
+    }
+
+    private sealed class FailingManifestStore(ISecretStore inner) : ISecretStore
+    {
+        public Task<StoredSecret?> GetAsync(SecretName name, CancellationToken ct) => inner.GetAsync(name, ct);
+        public Task UpsertAsync(StoredSecret row, CancellationToken ct) => inner.UpsertAsync(row, ct);
+        public Task<bool> DeleteAsync(SecretName name, string? actor, CancellationToken ct) => inner.DeleteAsync(name, actor, ct);
+        public Task<IReadOnlyList<SecretMetadata>> ListAsync(bool includeDeleted, CancellationToken ct) => inner.ListAsync(includeDeleted, ct);
+        public async Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct)
+        {
+            await Task.Delay(5, ct).ConfigureAwait(false);
+            throw new InvalidOperationException("manifest-read-boom");
+        }
+        public Task ApplyReplicatedAsync(StoredSecret row, CancellationToken ct) => inner.ApplyReplicatedAsync(row, ct);
+        public Task<bool> ApplyRewrapAsync(StoredSecret rewrappedRow, byte[] expectedCurrentWrappedDek, CancellationToken ct) =>
+            inner.ApplyRewrapAsync(rewrappedRow, expectedCurrentWrappedDek, ct);
+    }
+
+    // ---------- Secrets-011 ----------
+
+    [Fact]
+    public void An_announce_from_a_genuinely_async_store_carries_the_actor_as_sender()
+    {
+        SubscribeTestActor();
+        IActorRef node = Sys.ActorOf(
+            SecretReplicationActor.Props(new GenuinelyAsyncSecretStore(CreateStore()), null, AnnounceInterval),
+            "async-announcer");
+
+        ExpectMsg<SecretManifestAnnounce>(TimeSpan.FromSeconds(10));
+
+        // A peer replies to the announce's sender; deadLetters here means every reply is lost.
+        Assert.Equal(node, LastSender);
+    }
+
+    [Fact]
+    public void A_failed_manifest_read_is_logged_not_dropped()
+    {
+        EventFilter.Warning(contains: "Could not read the local secret manifest").ExpectOne(
+            TimeSpan.FromSeconds(10),
+            () => Sys.ActorOf(
+                SecretReplicationActor.Props(new FailingManifestStore(CreateStore()), null, AnnounceInterval),
+                "failing-announcer"));
+    }
+
+    // ---------- Secrets-012 ----------
+
+    private async Task<SqliteSecretStore> StoreWithAsync(int count, int ciphertextBytes)
+    {
+        SqliteSecretStore store = CreateStore();
+        for (int i = 0; i < count; i++)
+        {
+            await store.UpsertAsync(Row($"bulk/{i:D4}", ciphertextBytes), CancellationToken.None);
+        }
+
+        return store;
+    }
+
+    private List<SecretRowsMessage> ReceiveRows(int expectedRows)
+    {
+        var chunks = new List<SecretRowsMessage>();
+        int received = 0;
+        while (received < expectedRows)
+        {
+            SecretRowsMessage chunk = ExpectMsg<SecretRowsMessage>(TimeSpan.FromSeconds(10));
+            chunks.Add(chunk);
+            received += chunk.Rows.Count;
+        }
+
+        return chunks;
+    }
+
+    [Fact]
+    public async Task A_cold_peers_push_set_is_split_under_the_frame_size()
+    {
+        SqliteSecretStore store = await StoreWithAsync(300, ciphertextBytes: 600);
+        IActorRef node = Sys.ActorOf(SecretReplicationActor.Props(store, null, TimeSpan.FromHours(1)), "pusher");
+
+        // A cold peer announces an empty manifest: the node pushes its whole inventory.
+        node.Tell(new SecretManifestAnnounce([]), TestActor);
+
+        List<SecretRowsMessage> chunks = ReceiveRows(300);
+        Assert.True(chunks.Count > 1, "expected the push to be chunked");
+        Assert.Equal(300, chunks.Sum(c => c.Rows.Count));
+        Assert.All(chunks, c => Assert.True(SerializedSize(c) <= SecretReplicationChunks.MaxPayloadBytes));
+        Assert.Equal(300, chunks.SelectMany(c => c.Rows).Select(r => r.Name).Distinct().Count());
+    }
+
+    [Fact]
+    public async Task A_pull_reply_is_split_under_the_frame_size()
+    {
+        SqliteSecretStore store = await StoreWithAsync(300, ciphertextBytes: 600);
+        IActorRef node = Sys.ActorOf(SecretReplicationActor.Props(store, null, TimeSpan.FromHours(1)), "puller");
+
+        node.Tell(new SecretPullRequest([.. Enumerable.Range(0, 300).Select(i => $"bulk/{i:D4}")]), TestActor);
+
+        List<SecretRowsMessage> chunks = ReceiveRows(300);
+        Assert.True(chunks.Count > 1, "expected the pull reply to be chunked");
+        Assert.All(chunks, c => Assert.True(SerializedSize(c) <= SecretReplicationChunks.MaxPayloadBytes));
+    }
+
+    [Fact]
+    public void A_pull_request_for_a_large_inventory_is_split_under_the_frame_size()
+    {
+        IActorRef node = Sys.ActorOf(SecretReplicationActor.Props(CreateStore(), null, TimeSpan.FromHours(1)), "asker");
+        string longPrefix = new('n', 200);
+        DateTimeOffset now = DateTimeOffset.UtcNow;
+
+        // The peer holds 600 long names this (empty) node lacks: it must ask for them in pieces.
+        node.Tell(
+            new SecretManifestAnnounce(
+            [
+                .. Enumerable.Range(0, 600).Select(i => SecretManifestEntryDto.FromEntry(
+                    new SecretManifestEntry { Name = new SecretName($"{longPrefix}/{i:D4}"), UpdatedUtc = now, Revision = 1 })),
+            ]),
+            TestActor);
+
+        var requests = new List<SecretPullRequest>();
+        while (requests.Sum(r => r.Names.Count) < 600)
+        {
+            requests.Add(ExpectMsg<SecretPullRequest>(TimeSpan.FromSeconds(10)));
+        }
+
+        Assert.True(requests.Count > 1, "expected the pull request to be chunked");
+        Assert.All(requests, r => Assert.True(SerializedSize(r) <= SecretReplicationChunks.MaxPayloadBytes));
+    }
+
+    [Fact]
+    public void A_single_row_larger_than_the_budget_is_still_sent_alone()
+    {
+        var big = SecretRowDto.FromStoredSecret(Row("big/one", 80_000));
+        var small = SecretRowDto.FromStoredSecret(Row("small/one", 10));
+
+        List<SecretRowsMessage> chunks = [.. SecretReplicationChunks.Rows([small, big, small with { Name = "small/two" }])];
+
+        Assert.Equal(3, chunks.Count);
+        Assert.Equal("big/one", Assert.Single(chunks[1].Rows).Name);
+    }
+}