using ZB.MOM.WW.Secrets.Abstractions; using ZB.MOM.WW.Secrets.Replication; using ZB.MOM.WW.Secrets.Sqlite; using ZB.MOM.WW.Secrets.Tests.Fakes; namespace ZB.MOM.WW.Secrets.Tests.Replication; /// /// Convergence rules for the transport-agnostic reconciler. These are the cases both replication /// packages inherit, so they are pinned here once against a real SQLite store rather than a fake — /// a reconciler that agrees with a mock but not with the store would be worthless. /// public sealed class SecretReplicationReconcilerTests : IDisposable { private readonly string _dbPath = Path.Combine(Path.GetTempPath(), $"zb-recon-{Guid.NewGuid():N}.db"); private readonly SecretsSqliteConnectionFactory _factory; private readonly SqliteSecretStore _store; public SecretReplicationReconcilerTests() { _factory = new SecretsSqliteConnectionFactory(_dbPath); new SqliteSecretsStoreMigrator(_factory).MigrateAsync(CancellationToken.None).GetAwaiter().GetResult(); _store = new SqliteSecretStore(_factory); } private static readonly DateTimeOffset T0 = new(2026, 7, 18, 12, 0, 0, TimeSpan.Zero); private static SecretManifestEntry Entry(string name, long revision, DateTimeOffset updated, bool deleted = false) => new() { Name = new SecretName(name), Revision = revision, UpdatedUtc = updated, IsDeleted = deleted }; private static StoredSecret Row(string name, long revision, DateTimeOffset updated, bool deleted = false) => new() { Name = new SecretName(name), ContentType = SecretContentType.Text, Ciphertext = [1, 2, 3], Nonce = [4], Tag = [5], WrappedDek = [6], WrapNonce = [7], WrapTag = [8], KekId = "sha256:test", Revision = revision, IsDeleted = deleted, DeletedUtc = deleted ? updated : null, CreatedUtc = T0, UpdatedUtc = updated, }; [Fact] public void ComputePullSet_pulls_names_absent_locally() { IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [], remote: [Entry("db/password", 0, T0)]); Assert.Equal(["db/password"], pull.Select(n => n.Value)); } [Fact] public void ComputePullSet_pulls_a_newer_remote_row() { IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("api/key", 1, T0)], remote: [Entry("api/key", 2, T0.AddSeconds(1))]); Assert.Equal(["api/key"], pull.Select(n => n.Value)); } [Fact] public void ComputePullSet_breaks_a_timestamp_tie_on_revision() { IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("api/key", 1, T0)], remote: [Entry("api/key", 2, T0)]); Assert.Equal(["api/key"], pull.Select(n => n.Value)); } [Fact] public void ComputePullSet_ignores_an_identical_row_so_redelivery_is_free() { // At-least-once transports redeliver. A tie must not be "newer", or two nodes would // ping-pong the same row forever. IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("api/key", 2, T0)], remote: [Entry("api/key", 2, T0)]); Assert.Empty(pull); } [Fact] public void ComputePullSet_ignores_a_peer_that_is_behind() { IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("api/key", 5, T0.AddMinutes(1))], remote: [Entry("api/key", 2, T0)]); Assert.Empty(pull); } [Fact] public void ComputePullSet_pulls_a_tombstone_so_deletes_propagate() { // A delete is a newer row, not an absence — otherwise a deleted secret would be resurrected // by the next anti-entropy sweep from a node that still has it. IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("api/key", 1, T0)], remote: [Entry("api/key", 2, T0.AddSeconds(1), deleted: true)]); Assert.Equal(["api/key"], pull.Select(n => n.Value)); } [Fact] public void ComputePullSet_does_not_pull_a_name_only_the_local_side_has() { // The local side only ever pulls. Pushing local-only names is the peer's own reconcile. IReadOnlyList pull = SecretReplicationReconciler.ComputePullSet( localManifest: [Entry("local/only", 0, T0)], remote: []); Assert.Empty(pull); } [Fact] public async Task ReconcileAsync_applies_only_the_rows_the_peer_holds_newer() { await _store.UpsertAsync(Row("shared/a", 0, T0), CancellationToken.None); StoredSecret localA = (await _store.GetAsync(new SecretName("shared/a"), CancellationToken.None))!; var reconciler = new SecretReplicationReconciler(_store); List? requested = null; int applied = await reconciler.ReconcileAsync( remoteManifest: [ // Behind the local copy — must not be fetched. Entry("shared/a", localA.Revision, localA.UpdatedUtc.AddMinutes(-1)), // Absent locally — must be fetched. Entry("shared/b", 3, T0.AddHours(1)), ], fetchAsync: (names, _) => { requested = [.. names]; return Task.FromResult>([Row("shared/b", 3, T0.AddHours(1))]); }, CancellationToken.None); Assert.Equal(1, applied); Assert.Equal(["shared/b"], requested!.Select(n => n.Value)); StoredSecret? b = await _store.GetAsync(new SecretName("shared/b"), CancellationToken.None); Assert.NotNull(b); // Applied VERBATIM — the peer's revision, not a locally bumped one. Assert.Equal(3, b!.Revision); } [Fact] public async Task ReconcileAsync_skips_the_fetch_entirely_when_nothing_is_newer() { var reconciler = new SecretReplicationReconciler(_store); bool fetched = false; int applied = await reconciler.ReconcileAsync( remoteManifest: [], fetchAsync: (_, _) => { fetched = true; return Task.FromResult>([]); }, CancellationToken.None); Assert.Equal(0, applied); Assert.False(fetched); } [Fact] public async Task ApplyAsync_evicts_the_resolver_cache_for_every_row() { var invalidator = new RecordingCacheInvalidator(); var reconciler = new SecretReplicationReconciler(_store, invalidator); await reconciler.ApplyAsync([Row("a/one", 1, T0), Row("b/two", 1, T0)], CancellationToken.None); Assert.Equal(["a/one", "b/two"], invalidator.Invalidated.Select(n => n.Value)); } [Fact] public async Task ApplyAsync_evicts_even_when_the_store_rejected_the_row_as_stale() { // Serving a stale plaintext for the rest of the TTL is worse than a redundant re-read. await _store.UpsertAsync(Row("api/key", 0, T0), CancellationToken.None); var invalidator = new RecordingCacheInvalidator(); var reconciler = new SecretReplicationReconciler(_store, invalidator); await reconciler.ApplyAsync([Row("api/key", 0, T0.AddYears(-1))], CancellationToken.None); Assert.Equal(["api/key"], invalidator.Invalidated.Select(n => n.Value)); } [Fact] public async Task ApplyAsync_skips_a_poison_row_and_keeps_applying_the_rest() { // A batch is not a transaction. Without per-row isolation, one row a peer cannot supply // cleanly abandons every row after it — and since the pull set is recomputed identically // each sweep, in the same manifest order, the same row poisons the same batch forever. var failures = new List(); var reconciler = new SecretReplicationReconciler( _store, cacheInvalidator: null, onRowFailed: (name, _) => failures.Add(name)); // A null ciphertext cannot be persisted (NOT NULL column) — stands in for any bad row. StoredSecret poison = Row("bad/row", 1, T0) with { Ciphertext = null! }; int applied = await reconciler.ApplyAsync( [Row("good/first", 1, T0), poison, Row("good/second", 1, T0)], CancellationToken.None); Assert.Equal(2, applied); Assert.Equal(["bad/row"], failures.Select(n => n.Value)); Assert.NotNull(await _store.GetAsync(new SecretName("good/first"), CancellationToken.None)); Assert.NotNull(await _store.GetAsync(new SecretName("good/second"), CancellationToken.None)); } [Fact] public async Task ReconcileAsync_fetches_in_bounded_batches() { // An unbounded fetch breaks on real backends: SQL Server caps a command at 2100 parameters, // so a cold-starting node pulling a large hub's entire inventory would fail identically // every interval and never receive a single row. const int total = SecretReplicationReconciler.FetchBatchSize + 20; List remote = [.. Enumerable.Range(0, total) .Select(i => Entry($"bulk/{i}", 1, T0))]; var batchSizes = new List(); var reconciler = new SecretReplicationReconciler(_store); int applied = await reconciler.ReconcileAsync( remote, fetchAsync: (names, _) => { batchSizes.Add(names.Count); return Task.FromResult>( [.. names.Select(n => Row(n.Value, 1, T0))]); }, CancellationToken.None); Assert.Equal(total, applied); Assert.Equal(2, batchSizes.Count); Assert.All(batchSizes, size => Assert.True(size <= SecretReplicationReconciler.FetchBatchSize)); } public void Dispose() { Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); foreach (string path in new[] { _dbPath, _dbPath + "-wal", _dbPath + "-shm" }) { if (File.Exists(path)) { File.Delete(path); } } } }