diff --git a/ZB.MOM.WW.Secrets/Directory.Packages.props b/ZB.MOM.WW.Secrets/Directory.Packages.props
index 0a39b60..2c9a616 100644
--- a/ZB.MOM.WW.Secrets/Directory.Packages.props
+++ b/ZB.MOM.WW.Secrets/Directory.Packages.props
@@ -77,6 +77,11 @@
+
+
diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubContractPinTests.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubContractPinTests.cs
new file mode 100644
index 0000000..1c7142f
--- /dev/null
+++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubContractPinTests.cs
@@ -0,0 +1,69 @@
+using Google.Protobuf.Reflection;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+///
+/// Pins the secrets_hub.v1 service surface to its two read RPCs.
+///
+///
+///
+/// The narrow contract IS the enforcement. In this topology every update originates at
+/// central and a site node is a strict follower, so ISecretReplicationHub's third method —
+/// ApplyReplicatedAsync — is deliberately not mirrored onto the wire. With no write RPC
+/// there, a mis-wired or compromised site node has no way to push a row into central's store
+/// regardless of what its authorization happens to allow. That guarantee costs nothing to keep and
+/// is silently lost the moment a third RPC appears, which is why it is pinned here rather than left
+/// to the proto's comment.
+///
+///
+/// Read off the generated descriptor rather than the .proto text: the descriptor is what the server
+/// actually binds and the client actually dials, so a method that reached the wire some other way
+/// would still be caught.
+///
+///
+public sealed class SecretsHubContractPinTests
+{
+ private const string PullOnlyRationale =
+ "The secrets hub is PULL-ONLY by construction: updates originate at central only, and the " +
+ "absence of a write RPC is what enforces it — not authorization, which a mis-wired or " +
+ "compromised follower may hold more of than intended. Adding an RPC to this service " +
+ "retires that guarantee. If a write path is genuinely wanted, it needs its own design " +
+ "decision and its own review, not a passing test.";
+
+ [Fact]
+ public void The_hub_service_exposes_exactly_the_two_read_rpcs()
+ {
+ string[] rpcs = RpcNames();
+
+ Assert.True(
+ rpcs is ["GetManifest", "GetSecrets"],
+ $"secrets_hub.v1 SecretsHub must expose exactly [GetManifest, GetSecrets]; it exposes " +
+ $"[{string.Join(", ", rpcs)}]. {PullOnlyRationale}");
+ }
+
+ [Fact]
+ public void No_rpc_on_the_hub_service_streams_in_either_direction()
+ {
+ // A streaming RPC is a channel a follower can write into for as long as it stays open. The
+ // contract is unary today; that is a property worth pinning rather than rediscovering.
+ foreach (MethodDescriptor method in SecretsHub.Descriptor.Methods)
+ {
+ Assert.False(
+ method.IsClientStreaming || method.IsServerStreaming,
+ $"secrets_hub.v1 SecretsHub.{method.Name} is a streaming RPC. {PullOnlyRationale}");
+ }
+ }
+
+ [Fact]
+ public void The_service_keeps_its_package_qualified_identity()
+ {
+ // The full name is the wire address every deployed follower dials. Renaming the service or
+ // its package is a silent break for anything already running, so it is pinned alongside the
+ // method set.
+ Assert.Equal("zb.mom.ww.secrets.hub.v1.SecretsHub", SecretsHub.Descriptor.FullName);
+ }
+
+ private static string[] RpcNames() =>
+ [.. SecretsHub.Descriptor.Methods.Select(method => method.Name).Order(StringComparer.Ordinal)];
+}
diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubIntegrationTests.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubIntegrationTests.cs
new file mode 100644
index 0000000..ecf2c77
--- /dev/null
+++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubIntegrationTests.cs
@@ -0,0 +1,520 @@
+using System.Diagnostics;
+using Grpc.Core;
+using Grpc.Net.Client;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+///
+/// The gRPC secrets hub end to end: a central node hosting the real service through its own DI and
+/// interceptor, a site node running the real client and sweep through its own DI, and a real gRPC
+/// call between them.
+///
+///
+///
+/// Everything the unit suites necessarily substitute is real here — the registration extensions, the
+/// generated client and service, the fail-closed interceptor as the container actually activates it,
+/// protobuf encoding on both sides, and two real SQLite stores. What that buys is the class of defect
+/// no unit test can reach: pieces that are each correct and are never connected to one another.
+///
+///
+/// No credential in this file is a real one. The two token values are fabricated for these
+/// tests, and one of the tests exists to assert that neither ever reaches a log sink on either side.
+///
+///
+public sealed class SecretsHubIntegrationTests : IAsyncLifetime
+{
+ private const string ConfiguredToken = "central-hub-credential-7f3a";
+ private const string WrongToken = "impostor-credential-91cd";
+
+ private static readonly TimeSpan Patience = TimeSpan.FromSeconds(10);
+
+ private readonly TempSecretStore _hubStore = new("hub");
+ private readonly TempSecretStore _localStore = new("site");
+ private readonly List _disposables = [];
+
+ ///
+ public Task InitializeAsync() => Task.CompletedTask;
+
+ ///
+ public async Task DisposeAsync()
+ {
+ foreach (IAsyncDisposable disposable in _disposables)
+ {
+ await disposable.DisposeAsync();
+ }
+
+ _localStore.Dispose();
+ _hubStore.Dispose();
+ }
+
+ // ---- 1. end-to-end convergence ----------------------------------------------------------
+
+ [Fact]
+ public async Task One_sweep_carries_every_row_across_the_wire_byte_identically()
+ {
+ // The four rows between them cover every shape the wire has to keep distinguishable: a fully
+ // populated row with non-UTC offsets on both timestamps, a row whose every nullable field is
+ // null, a row whose nullable strings are deliberately EMPTY rather than absent, and a
+ // tombstone. Empty-versus-null is the one a plain proto3 string would silently collapse.
+ StoredSecret populated = Row("db/central", 0xA1) with
+ {
+ Description = "central connection string",
+ ContentType = SecretContentType.ConnectionString,
+ Revision = 7,
+ CreatedUtc = new DateTimeOffset(2026, 1, 2, 3, 4, 5, 678, TimeSpan.FromHours(-5)).AddTicks(9012),
+ UpdatedUtc = new DateTimeOffset(2026, 8, 7, 9, 10, 11, 121, TimeSpan.FromHours(5.5)).AddTicks(3456),
+ CreatedBy = "alice",
+ UpdatedBy = "bob",
+ };
+ StoredSecret nulls = Row("api/token", 0xB2);
+ StoredSecret empties = Row("svc/empties", 0xC3) with
+ {
+ Description = string.Empty,
+ CreatedBy = string.Empty,
+ UpdatedBy = string.Empty,
+ };
+ StoredSecret tombstone = Row("retired/legacy", 0xD4) with
+ {
+ IsDeleted = true,
+ DeletedUtc = new DateTimeOffset(2026, 8, 7, 9, 30, 0, TimeSpan.FromHours(9)),
+ Revision = 3,
+ UpdatedBy = "carol",
+ };
+
+ await SeedHubAsync(populated, nulls, empties, tombstone);
+
+ SecretsHubTestClient client = await ConnectAsync();
+
+ int pulled = await client.Sweeper.SweepAsync(CancellationToken.None);
+
+ Assert.Equal(4, pulled);
+
+ foreach (StoredSecret expected in new[] { populated, nulls, empties, tombstone })
+ {
+ StoredSecret hubRow = await RequireAsync(_hubStore.Store, expected.Name);
+ StoredSecret localRow = await RequireAsync(_localStore.Store, expected.Name);
+
+ AssertIdentical(hubRow, localRow);
+ }
+ }
+
+ [Fact]
+ public async Task Convergence_evicts_the_resolver_cache_for_exactly_the_rows_it_pulled()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1), Row("api/token", 0xB2));
+ await _localStore.Store.ApplyReplicatedAsync(Row("only/local", 0x11), CancellationToken.None);
+
+ SecretsHubTestClient client = await ConnectAsync();
+
+ await client.Sweeper.SweepAsync(CancellationToken.None);
+
+ Assert.Equal(
+ ["api/token", "db/central"],
+ client.Invalidated.Select(name => name.Value).Order(StringComparer.Ordinal));
+ }
+
+ [Fact]
+ public async Task A_converged_follower_pulls_nothing_on_the_next_sweep()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1));
+
+ SecretsHubTestClient client = await ConnectAsync();
+ await client.Sweeper.SweepAsync(CancellationToken.None);
+
+ Assert.Equal(0, await client.Sweeper.SweepAsync(CancellationToken.None));
+ }
+
+ // ---- 2. delete propagation --------------------------------------------------------------
+
+ [Fact]
+ public async Task A_hub_side_tombstone_replaces_a_live_local_row()
+ {
+ StoredSecret live = Row("api/token", 0xB2) with
+ {
+ Revision = 1,
+ CreatedUtc = DateTimeOffset.UtcNow.AddDays(-1),
+ UpdatedUtc = DateTimeOffset.UtcNow.AddDays(-1),
+ };
+
+ await SeedHubAsync(live);
+ await _localStore.Store.ApplyReplicatedAsync(live, CancellationToken.None);
+
+ SecretsHubTestClient client = await ConnectAsync();
+
+ // Nothing to do yet: both sides hold the identical live row.
+ Assert.Equal(0, await client.Sweeper.SweepAsync(CancellationToken.None));
+
+ // A real delete on the hub — higher revision, newer timestamp, tombstone flag set.
+ Assert.True(await _hubStore.Store.DeleteAsync(live.Name, "admin", CancellationToken.None));
+
+ Assert.Equal(1, await client.Sweeper.SweepAsync(CancellationToken.None));
+
+ StoredSecret hubRow = await RequireAsync(_hubStore.Store, live.Name);
+ StoredSecret localRow = await RequireAsync(_localStore.Store, live.Name);
+
+ Assert.True(localRow.IsDeleted);
+ Assert.NotNull(localRow.DeletedUtc);
+ AssertIdentical(hubRow, localRow);
+ }
+
+ // ---- 3. hub outage at boot, then recovery -----------------------------------------------
+
+ [Fact]
+ public async Task A_hub_that_is_down_at_boot_leaves_the_last_known_good_store_intact_and_converges_on_recovery()
+ {
+ StoredSecret lastKnownGood = Row("db/central", 0x77) with
+ {
+ Revision = 1,
+ CreatedUtc = DateTimeOffset.UtcNow.AddDays(-2),
+ UpdatedUtc = DateTimeOffset.UtcNow.AddDays(-2),
+ };
+ await _localStore.Store.ApplyReplicatedAsync(lastKnownGood, CancellationToken.None);
+
+ StoredSecret newer = lastKnownGood with
+ {
+ Ciphertext = [0x88],
+ Revision = 2,
+ UpdatedUtc = DateTimeOffset.UtcNow,
+ };
+ await SeedHubAsync(newer);
+
+ // The hub exists, but this node cannot reach it: the transport is disconnected before the
+ // node boots, which is the ordinary "central is down while a site restarts" case.
+ var transport = new SwitchableHttpHandler();
+ SecretsHubTestServer server = await StartHubAsync();
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, transport, ConfiguredToken));
+
+ await client.Sweeper.StartAsync(CancellationToken.None);
+
+ await WaitForAsync(
+ () => client.Logs.WarningsFrom().Count > 0,
+ "the startup sweep never reported the outage");
+
+ // The node did not crash, and it is still serving exactly what it had.
+ Assert.NotNull(client.Sweeper.ExecuteTask);
+ Assert.False(client.Sweeper.ExecuteTask!.IsFaulted);
+ AssertIdentical(lastKnownGood, await RequireAsync(_localStore.Store, lastKnownGood.Name));
+
+ // Central comes back.
+ transport.ConnectTo(server.CreateHandler());
+
+ Assert.Equal(1, await client.Sweeper.SweepAsync(CancellationToken.None));
+ AssertIdentical(
+ await RequireAsync(_hubStore.Store, newer.Name),
+ await RequireAsync(_localStore.Store, newer.Name));
+
+ await client.Sweeper.StopAsync(CancellationToken.None);
+ Assert.False(client.Sweeper.ExecuteTask!.IsFaulted);
+ }
+
+ [Fact]
+ public async Task An_outage_costs_exactly_one_warning_and_no_row_ever_disappears()
+ {
+ // The failure mode this guards is not "the sweep failed" — it is a hub that answers "I have
+ // nothing" being indistinguishable from a hub that does not answer, which would empty the
+ // follower's store. A sweep against a dead transport must move nothing at all.
+ StoredSecret held = Row("api/token", 0x55);
+ await _localStore.Store.ApplyReplicatedAsync(held, CancellationToken.None);
+
+ var transport = new SwitchableHttpHandler();
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, transport, ConfiguredToken));
+
+ await client.Sweeper.StartAsync(CancellationToken.None);
+ await WaitForAsync(
+ () => client.Logs.WarningsFrom().Count > 0,
+ "the startup sweep never reported the outage");
+ await client.Sweeper.StopAsync(CancellationToken.None);
+
+ Assert.Single(client.Logs.WarningsFrom());
+ AssertIdentical(held, await RequireAsync(_localStore.Store, held.Name));
+ }
+
+ [Fact]
+ public async Task A_transport_failure_reaches_the_raw_sweep_as_a_fault_rather_than_an_empty_answer()
+ {
+ // The containment above is only meaningful because the underlying call really does fail.
+ var transport = new SwitchableHttpHandler();
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(
+ _localStore.Store, transport, ConfiguredToken, syncOnStartup: false));
+
+ await Assert.ThrowsAsync(
+ () => client.Sweeper.SweepAsync(CancellationToken.None));
+ }
+
+ // ---- 4. authentication, fail-closed, over the wire ---------------------------------------
+
+ [Fact]
+ public async Task A_wrong_bearer_token_is_refused()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1));
+ SecretsHubTestServer server = await StartHubAsync();
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), WrongToken));
+
+ RpcException error = await Assert.ThrowsAsync(
+ () => client.Hub.GetManifestAsync(CancellationToken.None));
+
+ Assert.Equal(StatusCode.Unauthenticated, error.StatusCode);
+ Assert.Null(await _localStore.Store.GetAsync(new SecretName("db/central"), CancellationToken.None));
+ }
+
+ [Fact]
+ public async Task A_call_with_no_credential_at_all_is_refused_in_exactly_the_same_words()
+ {
+ // Uniform denial, asserted where it matters: across the wire, between two DIFFERENT failure
+ // modes. A caller that could tell "no token" from "wrong token" would have a free oracle.
+ SecretsHubTestServer server = await StartHubAsync();
+ using GrpcChannel channel = ChannelOver(server.CreateHandler());
+ var bare = new SecretsHub.SecretsHubClient(channel);
+
+ RpcException noHeader = await Assert.ThrowsAsync(
+ async () => await bare.GetManifestAsync(new GetManifestRequest()));
+
+ SecretsHubTestClient wrong = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), WrongToken));
+ RpcException wrongToken = await Assert.ThrowsAsync(
+ () => wrong.Hub.GetManifestAsync(CancellationToken.None));
+
+ Assert.Equal(StatusCode.Unauthenticated, noHeader.StatusCode);
+ Assert.Equal(wrongToken.StatusCode, noHeader.StatusCode);
+ Assert.Equal(wrongToken.Status.Detail, noHeader.Status.Detail);
+ }
+
+ [Fact]
+ public async Task A_hub_with_no_token_configured_refuses_even_a_well_formed_credential()
+ {
+ SecretsHubTestServer server = Track(
+ await SecretsHubTestServer.StartWithNoTokenConfiguredAsync(_hubStore.Store));
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), ConfiguredToken));
+
+ RpcException error = await Assert.ThrowsAsync(
+ () => client.Hub.GetManifestAsync(CancellationToken.None));
+
+ Assert.Equal(StatusCode.Unauthenticated, error.StatusCode);
+ }
+
+ [Fact]
+ public async Task An_unconfigured_hub_warns_once_however_many_callers_it_refuses()
+ {
+ // Once per process, not once per call — and the interceptor can only honour that if the
+ // container hands gRPC the singleton it registered. That wiring is exactly what a unit test
+ // constructing the interceptor by hand cannot check.
+ SecretsHubTestServer server = Track(
+ await SecretsHubTestServer.StartWithNoTokenConfiguredAsync(_hubStore.Store));
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), ConfiguredToken));
+
+ for (int attempt = 0; attempt < 3; attempt++)
+ {
+ await Assert.ThrowsAsync(
+ () => client.Hub.GetManifestAsync(CancellationToken.None));
+ }
+
+ CapturedLog warning = Assert.Single(server.Logs.WarningsFrom());
+ Assert.Contains("Secrets:GrpcHub:BearerToken", warning.Message, StringComparison.Ordinal);
+ }
+
+ [Fact]
+ public async Task No_log_line_on_either_side_carries_credential_material()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1));
+ SecretsHubTestServer server = await StartHubAsync();
+
+ SecretsHubTestClient accepted = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), ConfiguredToken));
+ await accepted.Sweeper.SweepAsync(CancellationToken.None);
+
+ SecretsHubTestClient refused = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), WrongToken));
+ await Assert.ThrowsAsync(
+ () => refused.Hub.GetManifestAsync(CancellationToken.None));
+
+ IEnumerable everything = server.Logs.Entries
+ .Concat(accepted.Logs.Entries)
+ .Concat(refused.Logs.Entries);
+
+ // Captured at Trace on both hosts, so this covers the most verbose setting a deployment
+ // could switch on — the configured token, the presented one, and the header that carried it.
+ Assert.NotEmpty(everything);
+ foreach (CapturedLog entry in everything)
+ {
+ Assert.DoesNotContain(ConfiguredToken, entry.AllText, StringComparison.OrdinalIgnoreCase);
+ Assert.DoesNotContain(WrongToken, entry.AllText, StringComparison.OrdinalIgnoreCase);
+ }
+ }
+
+ // ---- 5. the request cap, over the wire ---------------------------------------------------
+
+ [Fact]
+ public async Task A_fetch_over_the_configured_cap_is_refused_as_resource_exhausted()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1), Row("api/token", 0xB2), Row("svc/empties", 0xC3));
+ SecretsHubTestServer server = await StartHubAsync(maxNamesPerRequest: 2);
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), ConfiguredToken));
+
+ RpcException error = await Assert.ThrowsAsync(
+ () => client.Hub.GetManyAsync(
+ [new SecretName("db/central"), new SecretName("api/token"), new SecretName("svc/empties")],
+ CancellationToken.None));
+
+ Assert.Equal(StatusCode.ResourceExhausted, error.StatusCode);
+ }
+
+ [Fact]
+ public async Task A_fetch_exactly_at_the_cap_is_served()
+ {
+ await SeedHubAsync(Row("db/central", 0xA1), Row("api/token", 0xB2));
+ SecretsHubTestServer server = await StartHubAsync(maxNamesPerRequest: 2);
+ SecretsHubTestClient client = Track(
+ SecretsHubTestClient.Create(_localStore.Store, server.CreateHandler(), ConfiguredToken));
+
+ IReadOnlyList rows = await client.Hub.GetManyAsync(
+ [new SecretName("db/central"), new SecretName("api/token")], CancellationToken.None);
+
+ Assert.Equal(2, rows.Count);
+ }
+
+ // ---- 6. transport faults stay distinguishable from parse faults --------------------------
+
+ [Fact]
+ public async Task An_RpcException_from_the_call_invoker_escapes_the_hub_client_unchanged()
+ {
+ // The client catches PARSE failures per item and drops them; it must never catch a transport
+ // failure. If it did, an unreachable hub would be indistinguishable from an empty one, and
+ // the sweep would have no way to tell "central said nothing" from "central has nothing".
+ var thrown = new RpcException(new Status(StatusCode.Unavailable, "hub unreachable"));
+ var invoker = new FakeSecretsHubCallInvoker { FaultWith = () => thrown };
+ GrpcSecretsHubClient client = ClientOver(invoker);
+
+ RpcException fromManifest = await Assert.ThrowsAsync(
+ () => client.GetManifestAsync(CancellationToken.None));
+ RpcException fromFetch = await Assert.ThrowsAsync(
+ () => client.GetManyAsync([new SecretName("db/central")], CancellationToken.None));
+
+ Assert.Same(thrown, fromManifest);
+ Assert.Same(thrown, fromFetch);
+ }
+
+ // ---- helpers -----------------------------------------------------------------------------
+
+ private static GrpcSecretsHubClient ClientOver(CallInvoker invoker) => new(
+ new SecretsHub.SecretsHubClient(invoker),
+ Options.Create(new SecretsGrpcHubClientOptions
+ {
+ Endpoint = "http://central.test",
+ BearerToken = "unused-by-the-invoker",
+ }),
+ new RecordingLogger());
+
+ private static GrpcChannel ChannelOver(HttpMessageHandler handler) =>
+ GrpcChannel.ForAddress("http://central.test", new GrpcChannelOptions { HttpHandler = handler });
+
+ private async Task StartHubAsync(int? maxNamesPerRequest = null) =>
+ Track(await SecretsHubTestServer.StartAsync(
+ _hubStore.Store, ConfiguredToken, maxNamesPerRequest));
+
+ private async Task ConnectAsync()
+ {
+ SecretsHubTestServer server = await StartHubAsync();
+
+ return Track(SecretsHubTestClient.Create(
+ _localStore.Store, server.CreateHandler(), ConfiguredToken, syncOnStartup: false));
+ }
+
+ private T Track(T disposable) where T : IAsyncDisposable
+ {
+ _disposables.Add(disposable);
+ return disposable;
+ }
+
+ // Seeded through ApplyReplicatedAsync rather than UpsertAsync: the upsert path stamps its own
+ // timestamps and revision, which would erase the exact values these tests are about.
+ private async Task SeedHubAsync(params StoredSecret[] rows)
+ {
+ foreach (StoredSecret row in rows)
+ {
+ await _hubStore.Store.ApplyReplicatedAsync(row, CancellationToken.None);
+ }
+ }
+
+ private static async Task RequireAsync(ISecretStore store, SecretName name)
+ {
+ StoredSecret? row = await store.GetAsync(name, CancellationToken.None);
+
+ Assert.True(row is not null, $"expected the store to hold '{name.Value}'");
+ return row!;
+ }
+
+ private static async Task WaitForAsync(Func condition, string because)
+ {
+ var elapsed = Stopwatch.StartNew();
+
+ while (!condition() && elapsed.Elapsed < Patience)
+ {
+ await Task.Delay(10);
+ }
+
+ Assert.True(condition(), because);
+ }
+
+ ///
+ /// Field-by-field equality of two rows — every byte array, every timestamp, every nullable.
+ ///
+ ///
+ /// Written out rather than delegated to record equality on purpose:
+ /// holds byte[], which records compare by REFERENCE, so Assert.Equal(a, b) on the
+ /// whole row would pass only for the same instance and silently prove nothing about a row that
+ /// crossed a wire. The byte-identical claim in the live gate is exactly this assertion.
+ ///
+ private static void AssertIdentical(StoredSecret expected, StoredSecret actual)
+ {
+ Assert.Equal(expected.Name.Value, actual.Name.Value);
+ Assert.Equal(expected.Description, actual.Description);
+ Assert.Equal(expected.ContentType, actual.ContentType);
+ Assert.Equal(expected.Ciphertext, actual.Ciphertext);
+ Assert.Equal(expected.Nonce, actual.Nonce);
+ Assert.Equal(expected.Tag, actual.Tag);
+ Assert.Equal(expected.WrappedDek, actual.WrappedDek);
+ Assert.Equal(expected.WrapNonce, actual.WrapNonce);
+ Assert.Equal(expected.WrapTag, actual.WrapTag);
+ Assert.Equal(expected.KekId, actual.KekId);
+ Assert.Equal(expected.Revision, actual.Revision);
+ Assert.Equal(expected.IsDeleted, actual.IsDeleted);
+ Assert.Equal(expected.DeletedUtc, actual.DeletedUtc);
+ Assert.Equal(expected.CreatedUtc, actual.CreatedUtc);
+ Assert.Equal(expected.UpdatedUtc, actual.UpdatedUtc);
+ Assert.Equal(expected.CreatedBy, actual.CreatedBy);
+ Assert.Equal(expected.UpdatedBy, actual.UpdatedBy);
+
+ // Offset, not just instant: DateTimeOffset equality is instant-only, so a wire form that
+ // normalized every timestamp to UTC would satisfy the assertions above while quietly
+ // discarding the offset the store persisted.
+ Assert.Equal(expected.CreatedUtc.Offset, actual.CreatedUtc.Offset);
+ Assert.Equal(expected.UpdatedUtc.Offset, actual.UpdatedUtc.Offset);
+ Assert.Equal(expected.DeletedUtc?.Offset, actual.DeletedUtc?.Offset);
+ }
+
+ private static StoredSecret Row(string name, byte marker) => new()
+ {
+ Name = new SecretName(name),
+ ContentType = SecretContentType.Text,
+ Ciphertext = [marker],
+ Nonce = [1, 2],
+ Tag = [3, 4, 5],
+ WrappedDek = [6, 7, 8, 9],
+ WrapNonce = [10],
+ WrapTag = [11, 12],
+ KekId = "sha256:kek-under-test",
+ Revision = 0,
+ CreatedUtc = new DateTimeOffset(2026, 8, 7, 1, 2, 3, TimeSpan.Zero),
+ UpdatedUtc = new DateTimeOffset(2026, 8, 7, 1, 2, 3, TimeSpan.Zero),
+ };
+}
diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubTestHarness.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubTestHarness.cs
new file mode 100644
index 0000000..e87b64f
--- /dev/null
+++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/SecretsHubTestHarness.cs
@@ -0,0 +1,375 @@
+using System.Collections.Concurrent;
+using Grpc.Net.Client;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.TestHost;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.DependencyInjection;
+using ZB.MOM.WW.Secrets.Sqlite;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+/// One captured log entry: enough to assert what was said and to prove what was not.
+/// The logger category the entry was written to.
+/// The level it was written at.
+/// The formatted message.
+/// The exception's full text, or the empty string when there was none.
+internal sealed record CapturedLog(string Category, LogLevel Level, string Message, string Exception)
+{
+ /// Message and exception text together — everything this entry could leak.
+ public string AllText => Message + "\n" + Exception;
+}
+
+///
+/// An that records every entry written anywhere in a host, so a test
+/// can assert both what was logged and — the security half — what never appeared in any log line.
+///
+///
+/// Registered at deliberately: an assertion that no bearer token ever
+/// reaches a sink is only worth making against the most verbose setting a deployment could enable.
+///
+internal sealed class CapturingLoggerProvider : ILoggerProvider
+{
+ private readonly ConcurrentQueue _entries = new();
+
+ /// Everything logged so far, in order.
+ public IReadOnlyList Entries => [.. _entries];
+
+ /// Entries at written by .
+ /// The logging type whose category is matched.
+ /// The matching warnings, in order.
+ public IReadOnlyList WarningsFrom() =>
+ [.. _entries.Where(entry =>
+ entry.Level == LogLevel.Warning &&
+ string.Equals(entry.Category, typeof(T).FullName, StringComparison.Ordinal))];
+
+ ///
+ public ILogger CreateLogger(string categoryName) => new CapturingLogger(categoryName, _entries);
+
+ ///
+ public void Dispose()
+ {
+ // Nothing to release, and the entries deliberately outlive the host: assertions run after
+ // the host that produced them has been torn down.
+ }
+
+ private sealed class CapturingLogger(string category, ConcurrentQueue sink) : ILogger
+ {
+ public IDisposable? BeginScope(TState state) where TState : notnull => null;
+
+ public bool IsEnabled(LogLevel logLevel) => true;
+
+ public void Log(
+ LogLevel logLevel,
+ EventId eventId,
+ TState state,
+ Exception? exception,
+ Func formatter)
+ {
+ ArgumentNullException.ThrowIfNull(formatter);
+
+ sink.Enqueue(new CapturedLog(
+ category,
+ logLevel,
+ formatter(state, exception),
+ exception?.ToString() ?? string.Empty));
+ }
+ }
+}
+
+///
+/// An whose backing transport can be swapped at runtime, including
+/// to "nothing at all".
+///
+///
+/// This is how the hub-outage scenario is expressed without a real socket: with no inner handler the
+/// send fails the way a refused connection does, and installing one afterwards models central coming
+/// back. The channel, the generated client and the sweep are all the real ones and never learn that
+/// anything changed — which is the point, since the behaviour under test is that the follower keeps
+/// serving its last-known-good store and converges once the hub answers again.
+///
+internal sealed class SwitchableHttpHandler : HttpMessageHandler
+{
+ private volatile HttpMessageInvoker? _inner;
+
+ /// Points the handler at a live transport.
+ /// The handler to forward to.
+ public void ConnectTo(HttpMessageHandler handler) => _inner = new HttpMessageInvoker(handler);
+
+ /// Takes the transport away, so every send fails as an unreachable host does.
+ public void Disconnect() => _inner = null;
+
+ ///
+ protected override Task SendAsync(
+ HttpRequestMessage request, CancellationToken cancellationToken)
+ {
+ HttpMessageInvoker? inner = _inner;
+
+ return inner is null
+ ? throw new HttpRequestException("The secrets hub refused the connection.")
+ : inner.SendAsync(request, cancellationToken);
+ }
+}
+
+/// A real SQLite-backed in a temp file, deleted on dispose.
+///
+/// Real store rather than a dictionary, for the same reason the offline sweep suite uses one: the
+/// behaviour under test is last-writer-wins convergence, and a hand-rolled fake would re-implement
+/// the very logic the assertions exist to check.
+///
+internal sealed class TempSecretStore : IDisposable
+{
+ private readonly string _path;
+
+ /// Creates and migrates a fresh store.
+ /// A short label folded into the file name to aid debugging.
+ public TempSecretStore(string label)
+ {
+ _path = Path.Combine(Path.GetTempPath(), $"zb-grpc-e2e-{label}-{Guid.NewGuid():N}.db");
+ var factory = new SecretsSqliteConnectionFactory(_path);
+ new SqliteSecretsStoreMigrator(factory).MigrateAsync(CancellationToken.None).GetAwaiter().GetResult();
+ Store = new SqliteSecretStore(factory);
+ }
+
+ /// The store.
+ public SqliteSecretStore Store { get; }
+
+ ///
+ public void Dispose()
+ {
+ Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+
+ foreach (string path in new[] { _path, _path + "-wal", _path + "-shm" })
+ {
+ if (File.Exists(path))
+ {
+ File.Delete(path);
+ }
+ }
+ }
+}
+
+///
+/// The central node: a real in-process ASP.NET Core host serving
+/// through AddZbSecretsGrpcHub + MapZbSecretsHub, over
+/// .
+///
+///
+/// Everything the production composition root does happens here — options binding from
+/// configuration, the fail-closed interceptor registered as a singleton and attached to the service,
+/// routing, endpoint mapping — so a defect in the wiring shows up as a failing call rather than as a
+/// unit test that constructed the pieces by hand and never noticed they were never connected.
+///
+internal sealed class SecretsHubTestServer : IAsyncDisposable
+{
+ private readonly IHost _host;
+
+ private SecretsHubTestServer(IHost host, CapturingLoggerProvider logs)
+ {
+ _host = host;
+ Logs = logs;
+ }
+
+ /// Everything the hub host logged.
+ public CapturingLoggerProvider Logs { get; }
+
+ /// A handler that dials this server in process.
+ /// The handler.
+ public HttpMessageHandler CreateHandler() => _host.GetTestServer().CreateHandler();
+
+ ///
+ /// Starts a hub registered exactly as a central node registers it.
+ ///
+ /// The store the hub serves.
+ /// The shared bearer token followers must present.
+ /// Optional per-request name cap.
+ /// The running hub.
+ public static Task StartAsync(
+ ISecretStore store, string bearerToken, int? maxNamesPerRequest = null)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(bearerToken);
+
+ var settings = new Dictionary
+ {
+ ["Secrets:GrpcHub:BearerToken"] = bearerToken,
+ };
+
+ if (maxNamesPerRequest is int cap)
+ {
+ settings["Secrets:GrpcHub:MaxNamesPerRequest"] =
+ cap.ToString(System.Globalization.CultureInfo.InvariantCulture);
+ }
+
+ IConfiguration configuration = new ConfigurationBuilder()
+ .AddInMemoryCollection(settings)
+ .Build();
+
+ return StartAsync(store, services => services.AddZbSecretsGrpcHub(configuration));
+ }
+
+ ///
+ /// Starts a hub whose options carry no bearer token.
+ ///
+ ///
+ /// AddZbSecretsGrpcHub refuses this configuration at registration — which is the intended
+ /// first line of defence and is pinned elsewhere — so the pieces are wired here the way that
+ /// extension wires them, minus the eager validation. That is precisely the case the interceptor's
+ /// deny-all exists for: a host that built the options some other way. Without this path the
+ /// second layer could only ever be tested against a hand-constructed interceptor, never over the
+ /// wire.
+ ///
+ /// The store the hub would serve.
+ /// The running hub.
+ public static Task StartWithNoTokenConfiguredAsync(ISecretStore store) =>
+ StartAsync(store, services =>
+ {
+ services.Configure(options => options.BearerToken = string.Empty);
+ services.AddSingleton();
+ services.AddGrpc()
+ .AddServiceOptions(
+ options => options.Interceptors.Add());
+ });
+
+ private static async Task StartAsync(
+ ISecretStore store, Action registerHub)
+ {
+ var logs = new CapturingLoggerProvider();
+
+ IHost host = await new HostBuilder()
+ .ConfigureWebHost(web =>
+ {
+ web.UseTestServer();
+ web.ConfigureServices(services =>
+ {
+ services.AddLogging(builder =>
+ {
+ builder.SetMinimumLevel(LogLevel.Trace);
+ builder.AddProvider(logs);
+ });
+ services.AddRouting();
+ services.AddSingleton(store);
+ registerHub(services);
+ });
+ web.Configure(app =>
+ {
+ app.UseRouting();
+ app.UseEndpoints(endpoints => endpoints.MapZbSecretsHub());
+ });
+ })
+ .StartAsync();
+
+ return new SecretsHubTestServer(host, logs);
+ }
+
+ ///
+ public async ValueTask DisposeAsync()
+ {
+ try
+ {
+ await _host.StopAsync(TimeSpan.FromSeconds(5));
+ }
+ catch (Exception)
+ {
+ // Teardown only: a host that has already faulted must not mask the assertion that ran
+ // before it.
+ }
+
+ _host.Dispose();
+ }
+}
+
+///
+/// The site node: the package's own client registration, dialling the in-process hub.
+///
+///
+/// AddZbSecretsGrpcHubClient is called for real; the only substitution is a keyed
+/// registered before it, which the extension's
+/// TryAddKeyedSingleton then leaves standing. So the generated client, the reader seam, the
+/// bearer header, the per-call deadline and the hosted sweep are all the shipped ones.
+///
+internal sealed class SecretsHubTestClient : IAsyncDisposable
+{
+ private readonly ServiceProvider _provider;
+
+ private SecretsHubTestClient(
+ ServiceProvider provider, CapturingLoggerProvider logs, List invalidated)
+ {
+ _provider = provider;
+ Logs = logs;
+ Invalidated = invalidated;
+ Sweeper = provider.GetServices().OfType().Single();
+ Hub = provider.GetRequiredService();
+ }
+
+ /// Everything the follower logged.
+ public CapturingLoggerProvider Logs { get; }
+
+ /// The shipped pull-only sweep, resolved as a hosted service.
+ public GrpcSecretSyncService Sweeper { get; }
+
+ /// The shipped hub reader, for the calls a sweep does not make on its own.
+ public ISecretsHubReader Hub { get; }
+
+ /// Names the sweep evicted from the resolver cache.
+ public List Invalidated { get; }
+
+ /// Builds a follower over the supplied transport.
+ /// The node's local store.
+ /// The transport reaching the hub.
+ /// The credential this node presents.
+ /// Whether the hosted sweep runs immediately on start.
+ /// The follower.
+ public static SecretsHubTestClient Create(
+ ISecretStore local,
+ HttpMessageHandler handler,
+ string bearerToken,
+ bool syncOnStartup = true)
+ {
+ var logs = new CapturingLoggerProvider();
+
+ IConfiguration configuration = new ConfigurationBuilder()
+ .AddInMemoryCollection(new Dictionary
+ {
+ ["Secrets:GrpcHub:Endpoint"] = "http://central.test",
+ ["Secrets:GrpcHub:BearerToken"] = bearerToken,
+ // Long enough that the only sweeps in a test are the ones the test asks for.
+ ["Secrets:GrpcHub:SyncInterval"] = "01:00:00",
+ ["Secrets:GrpcHub:SyncOnStartup"] = syncOnStartup ? "true" : "false",
+ })
+ .Build();
+
+ var services = new ServiceCollection();
+ services.AddLogging(builder =>
+ {
+ builder.SetMinimumLevel(LogLevel.Trace);
+ builder.AddProvider(logs);
+ });
+ services.AddSingleton(local);
+
+ List invalidated = [];
+ services.AddSingleton(new DelegatingCacheInvalidator(invalidated.Add));
+
+ // Registered ahead of the extension so its TryAddKeyedSingleton is a no-op: the endpoint in
+ // configuration is never dialled, this in-process transport is.
+ services.AddKeyedSingleton(
+ SecretsGrpcHubClientExtensions.ChannelServiceKey,
+ (_, _) => GrpcChannel.ForAddress(
+ "http://central.test", new GrpcChannelOptions { HttpHandler = handler }));
+
+ services.AddZbSecretsGrpcHubClient(configuration);
+
+ return new SecretsHubTestClient(services.BuildServiceProvider(), logs, invalidated);
+ }
+
+ ///
+ public ValueTask DisposeAsync() => _provider.DisposeAsync();
+
+ private sealed class DelegatingCacheInvalidator(Action onInvalidate) : ISecretCacheInvalidator
+ {
+ public void Invalidate(SecretName name) => onInvalidate(name);
+ }
+}
diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/TestDoubles.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/TestDoubles.cs
index ff5d757..ed412a7 100644
--- a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/TestDoubles.cs
+++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/TestDoubles.cs
@@ -33,6 +33,17 @@ internal sealed class FakeSecretsHubCallInvoker : CallInvoker
/// The most recent GetSecrets request, so tests can assert what was asked for.
public GetSecretsRequest? LastSecretsRequest { get; private set; }
+ ///
+ /// When set, every call fails with this exception instead of returning a canned response —
+ /// the transport half of the client's two failure kinds.
+ ///
+ ///
+ /// A parse failure is contained per item; a transport failure must escape untouched, or an
+ /// unreachable hub becomes indistinguishable from an empty one. Modelled as a factory so a test
+ /// can hand back the same instance and assert identity, or a fresh one per call.
+ ///
+ public Func? FaultWith { get; set; }
+
/// Returns the Authorization header value seen on the call at the given index.
/// Zero-based call index.
/// The header value, or when the header was not sent.
@@ -53,6 +64,18 @@ internal sealed class FakeSecretsHubCallInvoker : CallInvoker
LastSecretsRequest = secretsRequest;
}
+ if (FaultWith is not null)
+ {
+ // Faulted on the response task rather than thrown synchronously, because that is how a
+ // real call reports a transport failure — the client has to let it out of an await.
+ return new AsyncUnaryCall(
+ Task.FromException(FaultWith()),
+ Task.FromResult(new Metadata()),
+ () => Status.DefaultSuccess,
+ () => new Metadata(),
+ () => { });
+ }
+
object response = method.Name switch
{
"GetManifest" => ManifestResponse,
diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests.csproj b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests.csproj
index 2979313..beb86e7 100644
--- a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests.csproj
+++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests.csproj
@@ -6,6 +6,7 @@
+