feat(secrets): cluster replication via SQL Server and Akka.NET (G-7, 0.2.0)
Secrets were per-node SQLite, so a secret written on one node was invisible to the rest of a cluster. G-7's design resolved the "shared SQL store vs Akka replicator" fork to build only the former; both are built here so the choice is a deployment decision (availability vs partition tolerance) rather than a library limitation. Two new packages — ZB.MOM.WW.Secrets.Replicator.SqlServer (shared store, plus a local-store-with-hub mode) and .Replicator.AkkaDotNet (peer-to-peer over distributed pub/sub). Core gains ISecretsStoreMigrator, one shared SecretLastWriterWins predicate so no two stores can disagree on a tie, the transport-agnostic reconciler, and ReplicatingSecretStore — which closes a real gap: nothing had ever called ISecretReplicator.PublishAsync, so the seam was inert and local writes would not have propagated at all. Verified 182 pass / 1 skip / 0 warnings, including 15 live tests against a real SQL Server 2022 (the SQLite suite ported case-for-case, so any behavioural divergence between the stores fails) and a 9-test in-process 2-node Akka cluster over real remoting. A post-build review caught six defects, all fixed and now covered: both replication modes could not resolve from the container (no test had built one), an unbounded fetch that broke past SQL Server's 2100-parameter cap, a poison row that aborted the rest of its batch forever, Enum.Parse on peer input that could restart the actor in a loop, null crypto blobs crossing the trust boundary, and a silently dropped pull-read failure. Packed at 0.2.0 and vulnerability-scanned clean; not yet published to the feed. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
+81
@@ -0,0 +1,81 @@
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
using ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Protocol;
|
||||
|
||||
namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// The DTO is the trust boundary for anything a peer sends. Every rejection here must surface as an
|
||||
/// <see cref="ArgumentException"/>, because that is the exception type the receiving actor filters
|
||||
/// on — anything else escapes its handler, restarts the actor, and (since the row is redelivered)
|
||||
/// loops.
|
||||
/// </summary>
|
||||
public sealed class HostileWireInputTests
|
||||
{
|
||||
private static SecretRowDto Valid() => new()
|
||||
{
|
||||
Name = "app/ok",
|
||||
ContentType = "Text",
|
||||
Ciphertext = [1],
|
||||
Nonce = [2],
|
||||
Tag = [3],
|
||||
WrappedDek = [4],
|
||||
WrapNonce = [5],
|
||||
WrapTag = [6],
|
||||
KekId = "sha256:x",
|
||||
Revision = 0,
|
||||
CreatedUtc = "2026-01-01T00:00:00.0000000+00:00",
|
||||
UpdatedUtc = "2026-01-01T00:00:00.0000000+00:00",
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public void A_valid_row_materializes()
|
||||
{
|
||||
StoredSecret row = Valid().ToStoredSecret();
|
||||
|
||||
Assert.Equal("app/ok", row.Name.Value);
|
||||
Assert.Equal(SecretContentType.Text, row.ContentType);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("../../etc/passwd")]
|
||||
[InlineData("/rooted")]
|
||||
[InlineData("has space")]
|
||||
[InlineData("")]
|
||||
public void A_malformed_name_is_rejected(string name)
|
||||
{
|
||||
Assert.Throws<ArgumentException>(() => (Valid() with { Name = name }).ToStoredSecret());
|
||||
}
|
||||
|
||||
[Theory]
|
||||
// An in-range number names no member — Enum.Parse would happily produce (SecretContentType)4096
|
||||
// and the store would persist it for later readers that assume a defined value.
|
||||
[InlineData("4096")]
|
||||
// Overflows the underlying type: Enum.Parse throws OverflowException, which is NOT an
|
||||
// ArgumentException, so it would escape the actor's filter and restart it.
|
||||
[InlineData("99999999999999999999")]
|
||||
[InlineData("NotAContentType")]
|
||||
[InlineData("")]
|
||||
public void An_unrecognized_content_type_is_rejected_as_an_ArgumentException(string contentType)
|
||||
{
|
||||
Assert.Throws<ArgumentException>(
|
||||
() => (Valid() with { ContentType = contentType }).ToStoredSecret());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_null_crypto_blob_is_rejected_at_the_boundary()
|
||||
{
|
||||
// `required byte[]` means "present in the payload", not "non-null" — an explicit JSON null
|
||||
// satisfies it. Caught here, or it blows up deep in the store's parameter binding instead.
|
||||
Assert.Throws<ArgumentException>(
|
||||
() => (Valid() with { Ciphertext = null! }).ToStoredSecret());
|
||||
Assert.Throws<ArgumentException>(
|
||||
() => (Valid() with { WrappedDek = null! }).ToStoredSecret());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_malformed_timestamp_is_rejected()
|
||||
{
|
||||
Assert.ThrowsAny<FormatException>(
|
||||
() => (Valid() with { UpdatedUtc = "not-a-date" }).ToStoredSecret());
|
||||
}
|
||||
}
|
||||
+160
@@ -0,0 +1,160 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Serialization;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
using ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Protocol;
|
||||
|
||||
namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Wire-contract tests. Secret ciphertext crosses the network as these bytes, so the round-trip is
|
||||
/// asserted field-by-field rather than by equality alone — a silently dropped crypto field would
|
||||
/// produce a row that stores fine and fails to decrypt much later, on another node.
|
||||
/// </summary>
|
||||
public sealed class SecretReplicationSerializerTests : TestKit
|
||||
{
|
||||
public SecretReplicationSerializerTests()
|
||||
: base(AkkaSecretsReplication.SerializationConfig)
|
||||
{
|
||||
}
|
||||
|
||||
private static StoredSecret Row(string name) => new()
|
||||
{
|
||||
Name = new SecretName(name),
|
||||
Description = "a description",
|
||||
ContentType = SecretContentType.ConnectionString,
|
||||
Ciphertext = [1, 2, 3, 250],
|
||||
Nonce = [4, 5, 6],
|
||||
Tag = [7, 8, 9],
|
||||
WrappedDek = [10, 11, 12],
|
||||
WrapNonce = [13, 14, 15],
|
||||
WrapTag = [16, 17, 18],
|
||||
KekId = "sha256:abc",
|
||||
Revision = 42,
|
||||
IsDeleted = true,
|
||||
DeletedUtc = new DateTimeOffset(2026, 3, 2, 1, 0, 0, TimeSpan.Zero),
|
||||
CreatedUtc = new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero),
|
||||
UpdatedUtc = new DateTimeOffset(2026, 2, 1, 0, 0, 0, TimeSpan.Zero),
|
||||
CreatedBy = "alice",
|
||||
UpdatedBy = "bob",
|
||||
};
|
||||
|
||||
private object RoundTrip(object message)
|
||||
{
|
||||
Serializer serializer = Sys.Serialization.FindSerializerFor(message);
|
||||
Assert.IsType<SecretReplicationSerializer>(serializer);
|
||||
|
||||
byte[] bytes = serializer.ToBinary(message);
|
||||
string manifest = ((SerializerWithStringManifest)serializer).Manifest(message);
|
||||
|
||||
return ((SerializerWithStringManifest)serializer).FromBinary(bytes, manifest);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SecretRowsMessage_round_trips_every_field()
|
||||
{
|
||||
StoredSecret original = Row("app/db-conn");
|
||||
var message = new SecretRowsMessage([SecretRowDto.FromStoredSecret(original)]);
|
||||
|
||||
var restored = (SecretRowsMessage)RoundTrip(message);
|
||||
StoredSecret result = restored.Rows.Single().ToStoredSecret();
|
||||
|
||||
Assert.Equal(original.Name.Value, result.Name.Value);
|
||||
Assert.Equal(original.Description, result.Description);
|
||||
Assert.Equal(original.ContentType, result.ContentType);
|
||||
Assert.Equal(original.Ciphertext, result.Ciphertext);
|
||||
Assert.Equal(original.Nonce, result.Nonce);
|
||||
Assert.Equal(original.Tag, result.Tag);
|
||||
Assert.Equal(original.WrappedDek, result.WrappedDek);
|
||||
Assert.Equal(original.WrapNonce, result.WrapNonce);
|
||||
Assert.Equal(original.WrapTag, result.WrapTag);
|
||||
Assert.Equal(original.KekId, result.KekId);
|
||||
Assert.Equal(original.Revision, result.Revision);
|
||||
Assert.Equal(original.IsDeleted, result.IsDeleted);
|
||||
Assert.Equal(original.DeletedUtc, result.DeletedUtc);
|
||||
Assert.Equal(original.CreatedUtc, result.CreatedUtc);
|
||||
Assert.Equal(original.UpdatedUtc, result.UpdatedUtc);
|
||||
Assert.Equal(original.CreatedBy, result.CreatedBy);
|
||||
Assert.Equal(original.UpdatedBy, result.UpdatedBy);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Timestamps_survive_the_wire_exactly()
|
||||
{
|
||||
// Last-writer-wins compares UpdatedUtc for EQUALITY to break revision ties. A serializer that
|
||||
// rounded to the millisecond would turn a tie into a spurious "newer" and two nodes would
|
||||
// then overwrite each other forever, so tick-exactness is a correctness requirement.
|
||||
StoredSecret original = Row("app/precise") with
|
||||
{
|
||||
UpdatedUtc = new DateTimeOffset(2026, 2, 1, 12, 34, 56, TimeSpan.Zero).AddTicks(1234567),
|
||||
};
|
||||
|
||||
var restored = (SecretRowsMessage)RoundTrip(
|
||||
new SecretRowsMessage([SecretRowDto.FromStoredSecret(original)]));
|
||||
|
||||
Assert.Equal(original.UpdatedUtc, restored.Rows.Single().ToStoredSecret().UpdatedUtc);
|
||||
Assert.Equal(original.UpdatedUtc.Ticks, restored.Rows.Single().ToStoredSecret().UpdatedUtc.Ticks);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ManifestAnnounce_round_trips()
|
||||
{
|
||||
var message = new SecretManifestAnnounce(
|
||||
[
|
||||
SecretManifestEntryDto.FromEntry(new SecretManifestEntry
|
||||
{
|
||||
Name = new SecretName("a/one"),
|
||||
Revision = 3,
|
||||
UpdatedUtc = new DateTimeOffset(2026, 5, 5, 5, 5, 5, TimeSpan.Zero),
|
||||
IsDeleted = true,
|
||||
}),
|
||||
]);
|
||||
|
||||
var restored = (SecretManifestAnnounce)RoundTrip(message);
|
||||
SecretManifestEntry entry = restored.Entries.Single().ToEntry();
|
||||
|
||||
Assert.Equal("a/one", entry.Name.Value);
|
||||
Assert.Equal(3, entry.Revision);
|
||||
Assert.True(entry.IsDeleted);
|
||||
Assert.Equal(new DateTimeOffset(2026, 5, 5, 5, 5, 5, TimeSpan.Zero), entry.UpdatedUtc);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PullRequest_round_trips()
|
||||
{
|
||||
var restored = (SecretPullRequest)RoundTrip(new SecretPullRequest(["a/one", "b/two"]));
|
||||
|
||||
Assert.Equal(["a/one", "b/two"], restored.Names);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_unknown_manifest_fails_loudly_rather_than_guessing()
|
||||
{
|
||||
// A mixed-version cluster must fail on an unrecognized payload, not mis-deserialize a secret.
|
||||
var serializer = (SerializerWithStringManifest)Sys.Serialization
|
||||
.FindSerializerFor(new SecretPullRequest([]));
|
||||
|
||||
Assert.Throws<System.Runtime.Serialization.SerializationException>(
|
||||
() => serializer.FromBinary([1, 2, 3], "zbs-from-the-future-v9"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_wire_contract_carries_no_field_that_could_hold_key_material()
|
||||
{
|
||||
// Structural guard on the trust boundary: the DTO's surface is fixed and reviewed. If someone
|
||||
// adds a property to SecretRowDto, this test fails and forces a deliberate look at whether
|
||||
// the new field is safe to put on the wire.
|
||||
string[] properties = [.. typeof(SecretRowDto)
|
||||
.GetProperties()
|
||||
.Select(p => p.Name)
|
||||
.Order(StringComparer.Ordinal)];
|
||||
|
||||
Assert.Equal(
|
||||
[
|
||||
"Ciphertext", "ContentType", "CreatedBy", "CreatedUtc", "DeletedUtc", "Description",
|
||||
"IsDeleted", "KekId", "Name", "Nonce", "Revision", "Tag", "UpdatedBy", "UpdatedUtc",
|
||||
"WrapNonce", "WrapTag", "WrappedDek",
|
||||
],
|
||||
properties);
|
||||
}
|
||||
}
|
||||
+354
@@ -0,0 +1,354 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Cluster;
|
||||
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.Sqlite;
|
||||
|
||||
namespace ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Two real cluster nodes, two real SQLite stores, real remoting over loopback — the gate the G-7
|
||||
/// design called for, run in-process so it belongs to the ordinary offline suite instead of needing
|
||||
/// a deployed rig.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Nothing here is faked: each node runs its own <see cref="ActorSystem"/> with Akka.Remote on its
|
||||
/// own port, they form a cluster, and rows travel through the actual serializer and the actual
|
||||
/// distributed pub/sub mediator. A mock-based test of this behaviour would prove almost nothing —
|
||||
/// the failure modes worth catching (self-echo, sender loss across the mediator, tombstones that do
|
||||
/// not propagate, a partition that never re-converges) only appear when messages really cross nodes.
|
||||
/// </remarks>
|
||||
public sealed class TwoNodeClusterReplicationTests : IAsyncLifetime
|
||||
{
|
||||
private readonly List<string> _dbPaths = [];
|
||||
private ActorSystem _systemA = null!;
|
||||
private ActorSystem _systemB = null!;
|
||||
private SqliteSecretStore _storeA = null!;
|
||||
private SqliteSecretStore _storeB = null!;
|
||||
private ISecretStore _writableA = null!;
|
||||
private IActorRef _replicatorB = null!;
|
||||
|
||||
// Short so anti-entropy assertions do not dominate the suite's runtime.
|
||||
private static readonly TimeSpan AnnounceInterval = TimeSpan.FromMilliseconds(300);
|
||||
private static readonly TimeSpan Patience = TimeSpan.FromSeconds(20);
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
_storeA = CreateStore();
|
||||
_storeB = CreateStore();
|
||||
|
||||
// Port 0 lets the OS assign free ports, so parallel test runs cannot collide.
|
||||
_systemA = ActorSystem.Create("zb-secrets-cluster", ClusterConfig(port: 0));
|
||||
int portA = ClusterPort(_systemA);
|
||||
|
||||
// Node A seeds itself; node B joins it.
|
||||
_systemA.Dispose();
|
||||
_systemA = ActorSystem.Create("zb-secrets-cluster", ClusterConfig(portA, seedPort: portA));
|
||||
_systemB = ActorSystem.Create("zb-secrets-cluster", ClusterConfig(port: 0, seedPort: portA));
|
||||
|
||||
await AwaitClusterUpAsync(_systemA, expectedMembers: 2);
|
||||
await AwaitClusterUpAsync(_systemB, expectedMembers: 2);
|
||||
|
||||
IActorRef replicatorA = _systemA.ActorOf(
|
||||
SecretReplicationActor.Props(_storeA, null, AnnounceInterval), "zb-secret-replication");
|
||||
_replicatorB = _systemB.ActorOf(
|
||||
SecretReplicationActor.Props(_storeB, null, AnnounceInterval), "zb-secret-replication");
|
||||
|
||||
_writableA = new ReplicatingSecretStore(
|
||||
_storeA, new AkkaSecretReplicator(replicatorA), NullLogger<ReplicatingSecretStore>.Instance);
|
||||
|
||||
// The mediator gossips subscriptions between nodes; publishing before both sides are
|
||||
// registered would silently drop the message.
|
||||
await AwaitBothSubscribedAsync();
|
||||
}
|
||||
|
||||
private SqliteSecretStore CreateStore()
|
||||
{
|
||||
string path = Path.Combine(Path.GetTempPath(), $"zb-akka-{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 Config ClusterConfig(int port, int? seedPort = null)
|
||||
{
|
||||
string seeds = seedPort is null
|
||||
? "[]"
|
||||
: $"[\"akka.tcp://zb-secrets-cluster@127.0.0.1:{seedPort}\"]";
|
||||
|
||||
return ConfigurationFactory.ParseString($$"""
|
||||
akka {
|
||||
loglevel = WARNING
|
||||
actor.provider = cluster
|
||||
remote.dot-netty.tcp {
|
||||
hostname = "127.0.0.1"
|
||||
public-hostname = "127.0.0.1"
|
||||
port = {{port}}
|
||||
}
|
||||
cluster {
|
||||
seed-nodes = {{seeds}}
|
||||
downing-provider-class = "Akka.Cluster.SBR.SplitBrainResolverProvider"
|
||||
}
|
||||
}
|
||||
""").WithFallback(AkkaSecretsReplication.SerializationConfig);
|
||||
}
|
||||
|
||||
private static int ClusterPort(ActorSystem system) =>
|
||||
Cluster.Get(system).SelfAddress.Port
|
||||
?? throw new InvalidOperationException("Cluster address has no port.");
|
||||
|
||||
private static async Task AwaitClusterUpAsync(ActorSystem system, int expectedMembers)
|
||||
{
|
||||
Cluster cluster = Cluster.Get(system);
|
||||
DateTime deadline = DateTime.UtcNow + Patience;
|
||||
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (cluster.State.Members.Count(m => m.Status == MemberStatus.Up) >= expectedMembers)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await Task.Delay(100);
|
||||
}
|
||||
|
||||
throw new TimeoutException(
|
||||
$"Cluster did not reach {expectedMembers} Up members within {Patience}.");
|
||||
}
|
||||
|
||||
// Writes a throwaway secret on A and waits for it to land on B. That round trip is the only
|
||||
// reliable signal that pub/sub subscriptions have gossiped across both nodes.
|
||||
private async Task AwaitBothSubscribedAsync()
|
||||
{
|
||||
var probe = new SecretName("warmup/probe");
|
||||
DateTime deadline = DateTime.UtcNow + Patience;
|
||||
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
await _writableA.UpsertAsync(Row("warmup/probe", 0x01), CancellationToken.None);
|
||||
|
||||
if (await WaitForAsync(_storeB, probe, _ => true, TimeSpan.FromMilliseconds(500)))
|
||||
{
|
||||
await _storeA.DeleteAsync(probe, "warmup", CancellationToken.None);
|
||||
await _storeB.DeleteAsync(probe, "warmup", CancellationToken.None);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
throw new TimeoutException("Nodes did not establish pub/sub subscriptions in time.");
|
||||
}
|
||||
|
||||
private static StoredSecret Row(string name, byte marker) => new()
|
||||
{
|
||||
Name = new SecretName(name),
|
||||
ContentType = SecretContentType.Text,
|
||||
Ciphertext = [marker],
|
||||
Nonce = [1],
|
||||
Tag = [2],
|
||||
WrappedDek = [3],
|
||||
WrapNonce = [4],
|
||||
WrapTag = [5],
|
||||
KekId = "sha256:shared-kek",
|
||||
Revision = 0,
|
||||
CreatedUtc = DateTimeOffset.UtcNow,
|
||||
UpdatedUtc = DateTimeOffset.UtcNow,
|
||||
};
|
||||
|
||||
private static async Task<bool> WaitForAsync(
|
||||
ISecretStore store, SecretName name, Func<StoredSecret, bool> predicate, TimeSpan? timeout = null)
|
||||
{
|
||||
DateTime deadline = DateTime.UtcNow + (timeout ?? Patience);
|
||||
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
StoredSecret? row = await store.GetAsync(name, CancellationToken.None);
|
||||
|
||||
if (row is not null && predicate(row))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
await Task.Delay(50);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_secret_written_on_node_A_becomes_resolvable_on_node_B()
|
||||
{
|
||||
var name = new SecretName("db/password");
|
||||
await _writableA.UpsertAsync(Row("db/password", 0xAA), CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeB, name, r => r.Ciphertext[0] == 0xAA),
|
||||
"Node B never received the secret written on node A.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_ciphertext_arrives_byte_identical()
|
||||
{
|
||||
var name = new SecretName("app/exact");
|
||||
await _writableA.UpsertAsync(Row("app/exact", 0x7F), CancellationToken.None);
|
||||
|
||||
Assert.True(await WaitForAsync(_storeB, name, _ => true));
|
||||
|
||||
StoredSecret onA = (await _storeA.GetAsync(name, CancellationToken.None))!;
|
||||
StoredSecret onB = (await _storeB.GetAsync(name, CancellationToken.None))!;
|
||||
|
||||
Assert.Equal(onA.Ciphertext, onB.Ciphertext);
|
||||
Assert.Equal(onA.WrappedDek, onB.WrappedDek);
|
||||
Assert.Equal(onA.KekId, onB.KekId);
|
||||
// Verbatim: the peer keeps the originating node's revision and timestamp, or the row would
|
||||
// look newer on B than on A and bounce back.
|
||||
Assert.Equal(onA.Revision, onB.Revision);
|
||||
Assert.Equal(onA.UpdatedUtc, onB.UpdatedUtc);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_delete_on_node_A_tombstones_the_row_on_node_B()
|
||||
{
|
||||
var name = new SecretName("app/doomed");
|
||||
await _writableA.UpsertAsync(Row("app/doomed", 0xBB), CancellationToken.None);
|
||||
Assert.True(await WaitForAsync(_storeB, name, _ => true));
|
||||
|
||||
await _writableA.DeleteAsync(name, "operator", CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeB, name, r => r.IsDeleted),
|
||||
"The delete never propagated to node B.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_update_overwrites_the_earlier_value_on_the_peer()
|
||||
{
|
||||
var name = new SecretName("app/rotating");
|
||||
await _writableA.UpsertAsync(Row("app/rotating", 0x01), CancellationToken.None);
|
||||
Assert.True(await WaitForAsync(_storeB, name, r => r.Ciphertext[0] == 0x01));
|
||||
|
||||
await _writableA.UpsertAsync(Row("app/rotating", 0x02), CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeB, name, r => r.Ciphertext[0] == 0x02 && r.Revision == 1),
|
||||
"Node B did not converge on the updated value.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Anti_entropy_repairs_a_write_that_never_got_broadcast()
|
||||
{
|
||||
// Simulates a dropped publish — the write goes straight to node A's store, bypassing the
|
||||
// replicator entirely, so the ONLY route to node B is the periodic manifest exchange. This is
|
||||
// the case that makes best-effort publishing safe.
|
||||
var name = new SecretName("app/missed-broadcast");
|
||||
await _storeA.UpsertAsync(Row("app/missed-broadcast", 0xCC), CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeB, name, r => r.Ciphertext[0] == 0xCC),
|
||||
"Anti-entropy failed to repair a write that was never broadcast.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Anti_entropy_pulls_in_the_other_direction_too()
|
||||
{
|
||||
// The mirror case: node B holds a row node A has never heard of. A cannot request a name it
|
||||
// does not know exists, so this only converges if the exchange pushes as well as pulls.
|
||||
var name = new SecretName("app/born-on-b");
|
||||
await _storeB.UpsertAsync(Row("app/born-on-b", 0xDD), CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeA, name, r => r.Ciphertext[0] == 0xDD),
|
||||
"Node A never learned about a secret that originated on node B.");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_node_that_joins_late_catches_up_on_everything()
|
||||
{
|
||||
// Partition-heal / cold-start: the whole reason to choose this transport over a shared store.
|
||||
await _writableA.UpsertAsync(Row("bulk/one", 0x11), CancellationToken.None);
|
||||
await _writableA.UpsertAsync(Row("bulk/two", 0x22), CancellationToken.None);
|
||||
await _writableA.UpsertAsync(Row("bulk/three", 0x33), CancellationToken.None);
|
||||
|
||||
SqliteSecretStore lateStore = CreateStore();
|
||||
_systemB.ActorOf(
|
||||
SecretReplicationActor.Props(lateStore, null, AnnounceInterval), "late-joiner");
|
||||
|
||||
Assert.True(await WaitForAsync(lateStore, new SecretName("bulk/one"), r => r.Ciphertext[0] == 0x11));
|
||||
Assert.True(await WaitForAsync(lateStore, new SecretName("bulk/two"), r => r.Ciphertext[0] == 0x22));
|
||||
Assert.True(await WaitForAsync(lateStore, new SecretName("bulk/three"), r => r.Ciphertext[0] == 0x33));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_broadcast_round_trip_does_not_drift_the_originating_nodes_revision()
|
||||
{
|
||||
// Note on what this does and does NOT prove: the revision would also stay put if the self-echo
|
||||
// filter were removed, because re-applying an identical row ties on (updated_utc, revision)
|
||||
// and last-writer-wins rejects a tie. So this covers the drift symptom via defence in depth,
|
||||
// not the filter itself. The filter is what stops the echo being forwarded onward, which
|
||||
// needs three nodes to observe — worth adding if a third node ever joins this rig.
|
||||
var name = new SecretName("app/no-echo");
|
||||
await _writableA.UpsertAsync(Row("app/no-echo", 0xEE), CancellationToken.None);
|
||||
Assert.True(await WaitForAsync(_storeB, name, _ => true));
|
||||
|
||||
await Task.Delay(AnnounceInterval * 4);
|
||||
|
||||
StoredSecret onA = (await _storeA.GetAsync(name, CancellationToken.None))!;
|
||||
Assert.Equal(0, onA.Revision);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_malformed_row_from_a_peer_is_discarded_without_stalling_replication()
|
||||
{
|
||||
// Peer input is untrusted: a path-traversing name must be dropped, and — critically — the
|
||||
// actor must keep serving everything else afterwards.
|
||||
_replicatorB.Tell(new Protocol.SecretRowsMessage(
|
||||
[
|
||||
new Protocol.SecretRowDto
|
||||
{
|
||||
Name = "../../etc/passwd",
|
||||
ContentType = "Text",
|
||||
Ciphertext = [1], Nonce = [1], Tag = [1],
|
||||
WrappedDek = [1], WrapNonce = [1], WrapTag = [1],
|
||||
KekId = "sha256:x",
|
||||
Revision = 0,
|
||||
CreatedUtc = DateTimeOffset.UtcNow.ToString("O"),
|
||||
UpdatedUtc = DateTimeOffset.UtcNow.ToString("O"),
|
||||
},
|
||||
]));
|
||||
|
||||
var name = new SecretName("app/after-malformed");
|
||||
await _writableA.UpsertAsync(Row("app/after-malformed", 0x5A), CancellationToken.None);
|
||||
|
||||
Assert.True(
|
||||
await WaitForAsync(_storeB, name, r => r.Ciphertext[0] == 0x5A),
|
||||
"Replication stalled after a malformed row was received.");
|
||||
}
|
||||
|
||||
public async Task DisposeAsync()
|
||||
{
|
||||
await CoordinatedShutdown.Get(_systemA).Run(CoordinatedShutdown.ClrExitReason.Instance);
|
||||
await CoordinatedShutdown.Get(_systemB).Run(CoordinatedShutdown.ClrExitReason.Instance);
|
||||
|
||||
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
|
||||
|
||||
foreach (string db in _dbPaths)
|
||||
{
|
||||
foreach (string path in new[] { db, db + "-wal", db + "-shm" })
|
||||
{
|
||||
try
|
||||
{
|
||||
if (File.Exists(path))
|
||||
{
|
||||
File.Delete(path);
|
||||
}
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
// Best-effort temp cleanup.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<IsPackable>false</IsPackable>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="coverlet.collector" />
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" />
|
||||
<PackageReference Include="xunit" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" />
|
||||
<PackageReference Include="Akka.TestKit.Xunit2" />
|
||||
<PackageReference Include="Akka.Remote" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Using Include="Xunit" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\src\ZB.MOM.WW.Secrets.Replicator.AkkaDotNet\ZB.MOM.WW.Secrets.Replicator.AkkaDotNet.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user