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);
}
}
}
}