diff --git a/ZB.MOM.WW.Secrets/docs/operations/clustered-secrets.md b/ZB.MOM.WW.Secrets/docs/operations/clustered-secrets.md index c9091dc..021d194 100644 --- a/ZB.MOM.WW.Secrets/docs/operations/clustered-secrets.md +++ b/ZB.MOM.WW.Secrets/docs/operations/clustered-secrets.md @@ -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 diff --git a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/Protocol/SecretReplicationChunks.cs b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/Protocol/SecretReplicationChunks.cs new file mode 100644 index 0000000..ef35f7d --- /dev/null +++ b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/Protocol/SecretReplicationChunks.cs @@ -0,0 +1,65 @@ +using System.Text.Json; + +namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Protocol; + +/// +/// 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 +/// or it exceeds the default +/// akka.remote.dot-netty.tcp.maximum-frame-size (128000 bytes) at a few hundred secrets, is +/// dropped by the transport, and is rebuilt identically every announce — so the node never converges. +/// +/// +/// Each message is sized by the serializer's own encoding (compact STJ JSON) to at most +/// , 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. +/// +internal static class SecretReplicationChunks +{ + /// Byte budget for one serialized message: half the default 128000-byte frame. + internal const int MaxPayloadBytes = 64_000; + + // {"Rows":[ ... ]} / {"Names":[ ... ]} plus slack. + private const int EnvelopeBytes = 32; + + /// Groups into messages of at most . + internal static IEnumerable Rows( + IEnumerable rows, int budgetBytes = MaxPayloadBytes) => + Chunk(rows, row => JsonSerializer.SerializeToUtf8Bytes(row).Length, budgetBytes) + .Select(chunk => new SecretRowsMessage(chunk)); + + /// Groups into pull requests of at most . + internal static IEnumerable PullRequests( + IEnumerable names, int budgetBytes = MaxPayloadBytes) => + Chunk(names, name => JsonSerializer.SerializeToUtf8Bytes(name).Length, budgetBytes) + .Select(chunk => new SecretPullRequest(chunk)); + + private static IEnumerable> Chunk(IEnumerable items, Func size, int budgetBytes) + { + var current = new List(); + 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(); + used = EnvelopeBytes; + } + + current.Add(item); + used += itemBytes; + } + + if (current.Count > 0) + { + yield return current; + } + } +} diff --git a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/SecretReplicationActor.cs b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/SecretReplicationActor.cs index a281a6a..f2d19c2 100644 --- a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/SecretReplicationActor.cs +++ b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet/SecretReplicationActor.cs @@ -82,6 +82,9 @@ public sealed class SecretReplicationActor : ReceiveActor, IWithTimers Receive(f => _log.Warning(f.Cause, "Secret anti-entropy exchange with a peer failed; retrying next announce.")); Receive(_ => { /* successful exchange, piped back to self — nothing to do. */ }); + Receive(HandleManifestReady); + Receive(f => + _log.Warning(f.Cause, "Could not read the local secret manifest to announce; retrying next interval.")); } /// Props for the replication actor. @@ -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 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 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 Manifest); + + private sealed record ManifestFailed(Exception Cause); + private sealed record ExchangeDone { public static readonly ExchangeDone Instance = new(); diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests/AnnounceAndChunkingTests.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests/AnnounceAndChunkingTests.cs new file mode 100644 index 0000000..e4a20e0 --- /dev/null +++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests/AnnounceAndChunkingTests.cs @@ -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; + +/// +/// 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. +/// +public sealed class AnnounceAndChunkingTests : TestKit, IDisposable +{ + private static readonly TimeSpan AnnounceInterval = TimeSpan.FromMilliseconds(200); + private readonly List _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(TimeSpan.FromSeconds(10)); + } + + private sealed class FailingManifestStore(ISecretStore inner) : ISecretStore + { + public Task GetAsync(SecretName name, CancellationToken ct) => inner.GetAsync(name, ct); + public Task UpsertAsync(StoredSecret row, CancellationToken ct) => inner.UpsertAsync(row, ct); + public Task DeleteAsync(SecretName name, string? actor, CancellationToken ct) => inner.DeleteAsync(name, actor, ct); + public Task> ListAsync(bool includeDeleted, CancellationToken ct) => inner.ListAsync(includeDeleted, ct); + public async Task> 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 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(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 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 ReceiveRows(int expectedRows) + { + var chunks = new List(); + int received = 0; + while (received < expectedRows) + { + SecretRowsMessage chunk = ExpectMsg(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 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 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(); + while (requests.Sum(r => r.Names.Count) < 600) + { + requests.Add(ExpectMsg(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 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); + } +}