fix(secrets): Grpc hub client — per-row parse isolation (review Critical) + real-client CallInvoker tests

A single malformed row or manifest entry poisoned the whole sweep, permanently. WireToDomain.ToRows/ToManifest materialized eagerly, so one unparseable element threw out of the fetch callback, past SecretReplicationReconciler's per-batch handling, into SweepSafelyAsync's whole-sweep catch. The follower's pull set is derived from the hub's manifest, so the identical bad element arrived in the identical batch on every interval and nothing ever converged — logged as "sync failed", indistinguishable from a hub outage. Isolation now matches the Akka sibling's split exactly: WireToDomain stays a strict per-item parser and loses its two batch entry points (their absence is pinned by a reflection test), while GrpcSecretsHubClient converts each item inside its own try — the same shape as SecretReplicationActor.Materialize. A dropped item costs one warning naming the secret and a reason category, and every well-formed sibling still converges. Only parse-shaped exceptions are caught (ArgumentException, FormatException); a defect still propagates. A row whose own name is unparseable is logged under a marker rather than echoing the untrusted string, and no ciphertext, DEK or credential reaches a log line. GrpcSecretsHubClient had zero coverage — the existing fake implements the ISecretsHubReader seam directly, bypassing the header, the deadline and the conversion, which are the only things that class does. It is now driven for real through a fake CallInvoker handed to the generated client. The stale doc comment claiming the reconciler isolated the failure is replaced by one describing the mechanism that actually exists, and Endpoint gains a note that a plain-http hub sends the bearer token cleartext (doc only, no validation change). Tests 52 -> 67 in the package suite; solution green, no new warnings. Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
2026-08-07 06:29:51 -04:00
parent c5ea3a7b59
commit 1c31e6bcb8
8 changed files with 649 additions and 73 deletions
@@ -2,6 +2,7 @@ using Grpc.Net.Client;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using ZB.MOM.WW.Secrets.Abstractions;
 using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
@@ -85,6 +86,7 @@ public static class SecretsGrpcHubClientExtensions
            new SecretsHub.SecretsHubClient(
                sp.GetRequiredKeyedService<GrpcChannel>(ChannelServiceKey)),
            sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>(),
+            sp.GetRequiredService<ILogger<GrpcSecretsHubClient>>(),
            sp.GetService<TimeProvider>()));

        services.AddHostedService<GrpcSecretSyncService>();
@@ -1,4 +1,5 @@
 using Grpc.Core;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using ZB.MOM.WW.Secrets.Abstractions;
 using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
@@ -29,7 +30,11 @@ internal interface ISecretsHubReader
    /// <returns>The hub's manifest entries.</returns>
    Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct);

-    /// <summary>Fetches several encrypted rows by name; names absent from the hub are omitted.</summary>
+    /// <summary>
+    /// Fetches several encrypted rows by name. Names absent from the hub are omitted, and so is any
+    /// row the follower cannot parse — an implementation returns the well-formed rows rather than
+    /// failing the batch.
+    /// </summary>
    /// <param name="names">The names to fetch.</param>
    /// <param name="ct">A token to cancel the operation.</param>
    /// <returns>The matching encrypted rows.</returns>
@@ -42,33 +47,60 @@ internal interface ISecretsHubReader
 /// <see cref="WireToDomain"/>.
 /// </summary>
 /// <remarks>
-/// Faults are deliberately <b>not</b> caught here. An <see cref="RpcException"/> is the honest signal
-/// that the hub did not answer, and the sweep — the only caller — is where that is turned into "keep
-/// serving the last-known-good local store and retry next tick". Swallowing it here would make an
-/// unreachable hub indistinguishable from an empty one, which is the one failure mode this
+/// <para>
+/// <b>Transport faults are deliberately not caught.</b> An <see cref="RpcException"/> is the honest
+/// signal that the hub did not answer, and the sweep — the only caller — is where that is turned
+/// into "keep serving the last-known-good local store and retry next tick". Swallowing it here would
+/// make an unreachable hub indistinguishable from an empty one, which is the one failure mode this
 /// subsystem must never have.
+/// </para>
+/// <para>
+/// <b>Parse faults are caught, per item.</b> That is not the same decision: the hub answered, and
+/// the answer is mostly good. Failing the whole response over one unparseable element would stall
+/// convergence permanently rather than transiently — the follower's pull set comes from the hub's
+/// manifest, so the identical bad element would arrive in the identical batch on every interval and
+/// nothing would ever be applied, while the log said only "sync failed" exactly as it does for an
+/// outage. So each item is converted inside its own <c>try</c>, one warning names what was dropped
+/// and why, and every well-formed sibling is returned. This is the same containment the Akka
+/// replicator applies in <c>SecretReplicationActor.Materialize</c>.
+/// </para>
+/// <para>
+/// A dropped item is dropped whole and never coerced — the fail-closed half of the same rule. Only
+/// parse-shaped exceptions are caught (<see cref="ArgumentException"/>, <see cref="FormatException"/>);
+/// anything else is a defect rather than bad input and is left to propagate.
+/// </para>
 /// </remarks>
-internal sealed class GrpcSecretsHubClient : ISecretsHubReader
+internal sealed partial class GrpcSecretsHubClient : ISecretsHubReader
 {
    private const string AuthorizationHeader = "Authorization";

+    /// <summary>
+    /// Stands in for a name in a log line when the name itself is what failed to parse. The hub's
+    /// raw string is deliberately not echoed: it is the untrusted value under suspicion.
+    /// </summary>
+    private const string UnnameableItem = "(unparseable name)";
+
    private readonly SecretsHub.SecretsHubClient _client;
    private readonly SecretsGrpcHubClientOptions _options;
+    private readonly ILogger<GrpcSecretsHubClient> _logger;
    private readonly TimeProvider _timeProvider;

    /// <summary>Creates the hub client.</summary>
    /// <param name="client">The generated gRPC client, bound to the configured channel.</param>
    /// <param name="options">Endpoint, credential, and per-call deadline.</param>
+    /// <param name="logger">Logger for items the hub sent that this node could not parse.</param>
    /// <param name="timeProvider">Clock used to compute the per-call deadline.</param>
    public GrpcSecretsHubClient(
        SecretsHub.SecretsHubClient client,
        IOptions<SecretsGrpcHubClientOptions> options,
+        ILogger<GrpcSecretsHubClient> logger,
        TimeProvider? timeProvider = null)
    {
        ArgumentNullException.ThrowIfNull(options);

        _client = client ?? throw new ArgumentNullException(nameof(client));
        _options = options.Value;
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _timeProvider = timeProvider ?? TimeProvider.System;
    }

@@ -79,7 +111,23 @@ internal sealed class GrpcSecretsHubClient : ISecretsHubReader
            .GetManifestAsync(new GetManifestRequest(), CreateCallOptions(ct))
            .ConfigureAwait(false);

-        return WireToDomain.ToManifest(response.Entries);
+        var entries = new List<SecretManifestEntry>(response.Entries.Count);
+
+        foreach (ManifestEntry wire in response.Entries)
+        {
+            try
+            {
+                entries.Add(WireToDomain.ToEntry(wire));
+            }
+            catch (Exception ex) when (IsParseFailure(ex))
+            {
+                // Dropping a manifest entry costs only that name's convergence: it is simply not in
+                // this sweep's pull set, and a corrected entry converges on the next one.
+                LogManifestEntryDiscarded(ex, Describe(wire.Name), Categorize(ex));
+            }
+        }
+
+        return entries;
    }

    /// <inheritdoc />
@@ -99,7 +147,23 @@ internal sealed class GrpcSecretsHubClient : ISecretsHubReader
            .GetSecretsAsync(request, CreateCallOptions(ct))
            .ConfigureAwait(false);

-        return WireToDomain.ToRows(response.Secrets);
+        var rows = new List<StoredSecret>(response.Secrets.Count);
+
+        foreach (SecretRow wire in response.Secrets)
+        {
+            try
+            {
+                rows.Add(WireToDomain.ToRow(wire));
+            }
+            catch (Exception ex) when (IsParseFailure(ex))
+            {
+                // The reconciler treats a missing row exactly as it treats a name the hub no longer
+                // holds, so the batch applies its siblings and returns a smaller count.
+                LogRowDiscarded(ex, Describe(wire.Name), Categorize(ex));
+            }
+        }
+
+        return rows;
    }

    // Built per call rather than once: the deadline is an absolute instant, so a cached CallOptions
@@ -108,4 +172,46 @@ internal sealed class GrpcSecretsHubClient : ISecretsHubReader
        headers: new Metadata { { AuthorizationHeader, $"Bearer {_options.BearerToken}" } },
        deadline: _timeProvider.GetUtcNow().UtcDateTime.Add(_options.CallDeadline),
        cancellationToken: ct);
+
+    // The exceptions WireToDomain raises for input the domain refuses: SecretName and the content
+    // type check throw ArgumentException, the round-trip timestamp parse throws FormatException.
+    // Narrow on purpose — a NullReferenceException or an OutOfMemoryException is a defect or a
+    // resource failure, and silently dropping rows over either would hide it behind a warning.
+    private static bool IsParseFailure(Exception exception) =>
+        exception is ArgumentException or FormatException;
+
+    // The name is the only field of an item that may reach a log sink, and only once the domain has
+    // accepted it. Nothing else is logged: a row is ciphertext, a wrapped DEK and two nonces, and a
+    // warning is not a place any of that belongs.
+    private static string Describe(string name)
+    {
+        try
+        {
+            return new SecretName(name).Value;
+        }
+        catch (ArgumentException)
+        {
+            return UnnameableItem;
+        }
+    }
+
+    // A category rather than a restatement of the exception, so an operator reading the warning
+    // knows which authored field to go and look at.
+    private static string Categorize(Exception exception) => exception switch
+    {
+        FormatException => "malformed timestamp",
+        _ => "invalid name or content type",
+    };
+
+    [LoggerMessage(
+        EventId = 1,
+        Level = LogLevel.Warning,
+        Message = "Discarding a malformed manifest entry for {SecretName} sent by the secrets hub ({Reason}); every other entry is reconciled normally.")]
+    private partial void LogManifestEntryDiscarded(Exception exception, string secretName, string reason);
+
+    [LoggerMessage(
+        EventId = 2,
+        Level = LogLevel.Warning,
+        Message = "Discarding a malformed secret row for {SecretName} sent by the secrets hub ({Reason}); every other row in the batch is applied normally.")]
+    private partial void LogRowDiscarded(Exception exception, string secretName, string reason);
 }
@@ -17,6 +17,15 @@ public sealed class SecretsGrpcHubClientOptions
    /// The hub's gRPC address, as an absolute <c>http</c> or <c>https</c> URI (for example
    /// <c>https://central:5001</c>). Required.
    /// </summary>
+    /// <remarks>
+    /// A plain <c>http</c> endpoint is accepted so an h2c-only or in-cluster deployment can be
+    /// stood up, but it sends <see cref="BearerToken"/> in cleartext on <em>every</em> call — anyone
+    /// on the path reads the hub credential and can then pull central's entire secret inventory.
+    /// Treat <c>http</c> as a development and loopback-only convenience; a site node reaching
+    /// central across a plant network must use <c>https</c>. This is documentation rather than a
+    /// rule the validator enforces, because "internal" is a property of a deployment's network and
+    /// cannot be decided from a URI string.
+    /// </remarks>
    public string Endpoint { get; set; } = string.Empty;

    /// <summary>
@@ -10,16 +10,24 @@ namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Wire;
 /// </summary>
 /// <remarks>
 /// <para>
-/// Parsing here is <b>strict, and rejection is per row</b>: a value protobuf accepted but the domain
+/// Parsing here is <b>strict and one item at a time</b>: a value protobuf accepted but the domain
 /// would not — a name that is rooted or path-traversing, a content type naming no enum member, a
-/// timestamp that is not round-trip ISO-8601 — throws rather than being coerced. The sweep's
-/// reconciler isolates the failure to that one row (see <c>SecretReplicationReconciler.ApplyAsync</c>),
-/// so a single bad row is logged and skipped instead of poisoning the batch behind it.
+/// timestamp that is not round-trip ISO-8601 — throws rather than being coerced.
 /// </para>
 /// <para>
-/// This mirrors the Akka replicator's <c>SecretRowDto.ToStoredSecret</c> deliberately — one
-/// fail-closed parse shape family-wide, so the two transports cannot disagree about what a
-/// replicated row is allowed to contain.
+/// There is deliberately <b>no batch entry point here</b>, and that absence is the design. A method
+/// converting a whole response would fail the whole response on one bad element, and because a
+/// follower's pull set is derived from the hub's manifest, the identical bad element would re-fail
+/// the identical sweep on every interval forever — a permanent stall that reads in the log exactly
+/// like a hub outage. Isolation therefore lives one layer up, in
+/// <c>GrpcSecretsHubClient</c>, which converts each item inside its own <c>try</c>, logs one warning
+/// naming the item, drops it, and returns every well-formed sibling.
+/// </para>
+/// <para>
+/// That split mirrors the Akka replicator exactly — <c>SecretRowDto.ToStoredSecret</c> throws and
+/// <c>SecretReplicationActor.Materialize</c> catches per row — so the two transports share one
+/// fail-closed parse shape and one containment rule, and cannot disagree about what a replicated row
+/// is allowed to contain.
 /// </para>
 /// <para>
 /// Two checks the Akka DTO makes are <em>absent</em> here on purpose. Protobuf cannot express a null
@@ -32,16 +40,6 @@ namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Wire;
 /// </remarks>
 internal static class WireToDomain
 {
-    /// <summary>Rebuilds a manifest from the wire form, in the order the hub sent it.</summary>
-    /// <param name="entries">The wire manifest entries.</param>
-    /// <returns>The domain manifest entries.</returns>
-    public static IReadOnlyList<SecretManifestEntry> ToManifest(IEnumerable<ManifestEntry> entries)
-    {
-        ArgumentNullException.ThrowIfNull(entries);
-
-        return [.. entries.Select(ToEntry)];
-    }
-
    /// <summary>Rebuilds one manifest entry from the wire form.</summary>
    /// <param name="entry">The wire manifest entry.</param>
    /// <returns>The domain manifest entry.</returns>
@@ -62,16 +60,6 @@ internal static class WireToDomain
        };
    }

-    /// <summary>Rebuilds encrypted rows from the wire form, in the order the hub sent them.</summary>
-    /// <param name="rows">The wire rows.</param>
-    /// <returns>The domain rows.</returns>
-    public static IReadOnlyList<StoredSecret> ToRows(IEnumerable<SecretRow> rows)
-    {
-        ArgumentNullException.ThrowIfNull(rows);
-
-        return [.. rows.Select(ToRow)];
-    }
-
    /// <summary>Rebuilds one encrypted row from the wire form.</summary>
    /// <param name="row">The wire row.</param>
    /// <returns>The domain row.</returns>
@@ -7,6 +7,7 @@ using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using ZB.MOM.WW.Secrets.Abstractions;
 using ZB.MOM.WW.Secrets.Replicator.Grpc.DependencyInjection;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
 using ZB.MOM.WW.Secrets.Sqlite;

 namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
@@ -21,7 +22,8 @@ public sealed class GrpcSecretSyncServiceTests : IDisposable
    private readonly SqliteSecretStore _local;
    private readonly SqliteSecretStore _hubStore;
    private readonly FakeHubReader _hub;
-    private readonly RecordingLogger _logger = new();
+    private readonly RecordingLogger<GrpcSecretSyncService> _logger = new();
+    private readonly RecordingLogger<GrpcSecretsHubClient> _clientLogger = new();
    private readonly RecordingCacheInvalidator _cache = new();

    public GrpcSecretSyncServiceTests()
@@ -331,6 +333,81 @@ public sealed class GrpcSecretSyncServiceTests : IDisposable
        Assert.Equal(0, _hub.ManifestCallCount);
    }

+    [Fact]
+    public async Task A_malformed_row_never_poisons_the_batch_it_arrived_in()
+    {
+        // The failure this guards against is not "one row is lost" — it is that the pull set never
+        // changes, so the identical malformed row re-fails the identical sweep every interval and
+        // the node never converges on anything, indistinguishable in the log from a hub outage.
+        SecretRow bad = TestWire.Row("api/key", 0xBB);
+        bad.ContentType = "Kerberos";
+
+        var invoker = new FakeSecretsHubCallInvoker
+        {
+            ManifestResponse = new GetManifestResponse
+            {
+                Entries =
+                {
+                    TestWire.Entry("db/password"), TestWire.Entry("api/key"), TestWire.Entry("svc/token"),
+                },
+            },
+            SecretsResponse = new GetSecretsResponse
+            {
+                Secrets = { TestWire.Row("db/password", 0xAA), bad, TestWire.Row("svc/token", 0xCC) },
+            },
+        };
+
+        int pulled = await CreateServiceOverTheWire(invoker).SweepAsync(CancellationToken.None);
+
+        Assert.Equal(2, pulled);
+        Assert.NotNull(await _local.GetAsync(new SecretName("db/password"), CancellationToken.None));
+        Assert.NotNull(await _local.GetAsync(new SecretName("svc/token"), CancellationToken.None));
+        Assert.Null(await _local.GetAsync(new SecretName("api/key"), CancellationToken.None));
+        Assert.Single(_clientLogger.Warnings);
+    }
+
+    [Fact]
+    public async Task A_malformed_manifest_entry_never_poisons_the_manifest_it_arrived_in()
+    {
+        ManifestEntry bad = TestWire.Entry("api/key");
+        bad.UpdatedUtc = "not-a-timestamp";
+
+        var invoker = new FakeSecretsHubCallInvoker
+        {
+            ManifestResponse = new GetManifestResponse
+            {
+                Entries = { TestWire.Entry("db/password"), bad, TestWire.Entry("svc/token") },
+            },
+            SecretsResponse = new GetSecretsResponse
+            {
+                Secrets = { TestWire.Row("db/password", 0xAA), TestWire.Row("svc/token", 0xCC) },
+            },
+        };
+
+        int pulled = await CreateServiceOverTheWire(invoker).SweepAsync(CancellationToken.None);
+
+        Assert.Equal(2, pulled);
+        Assert.Equal(["db/password", "svc/token"], invoker.LastSecretsRequest!.Names);
+        Assert.Single(_clientLogger.Warnings);
+    }
+
+    // A sweep wired to the REAL hub client over a fake CallInvoker, so the parse boundary the other
+    // tests bypass (the FakeHubReader hands over domain types already) is actually exercised.
+    private GrpcSecretSyncService CreateServiceOverTheWire(FakeSecretsHubCallInvoker invoker)
+    {
+        IOptions<SecretsGrpcHubClientOptions> options = Options.Create(new SecretsGrpcHubClientOptions
+        {
+            Endpoint = "https://central:5001",
+            BearerToken = "hub-token",
+            SyncInterval = TimeSpan.FromHours(1),
+        });
+
+        var client = new GrpcSecretsHubClient(
+            new SecretsHub.SecretsHubClient(invoker), options, _clientLogger);
+
+        return new GrpcSecretSyncService(_local, client, _cache, options, _logger);
+    }
+
    [Fact]
    public void Registration_refuses_a_configuration_with_no_bearer_token()
    {
@@ -471,34 +548,4 @@ public sealed class GrpcSecretSyncServiceTests : IDisposable
        public void Invalidate(SecretName name) => Invalidated.Add(name);
    }

-    private sealed class RecordingLogger : ILogger<GrpcSecretSyncService>
-    {
-        public int InformationCount { get; private set; }
-
-        public int WarningCount { get; private set; }
-
-        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
-
-        public bool IsEnabled(LogLevel logLevel) => true;
-
-        public void Log<TState>(
-            LogLevel logLevel,
-            EventId eventId,
-            TState state,
-            Exception? exception,
-            Func<TState, Exception?, string> formatter)
-        {
-            switch (logLevel)
-            {
-                case LogLevel.Information:
-                    InformationCount++;
-                    break;
-                case LogLevel.Warning:
-                    WarningCount++;
-                    break;
-                default:
-                    break;
-            }
-        }
-    }
 }
@@ -0,0 +1,233 @@
+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;
+
+/// <summary>
+/// The real hub client, driven over a fake <c>CallInvoker</c>: what it puts on the wire (credential
+/// and deadline) and what it does with what comes back (per-item conversion, per-item isolation).
+/// </summary>
+/// <remarks>
+/// Everything here is deliberately exercised through <c>GrpcSecretsHubClient</c> itself rather than
+/// through a fake of the <c>ISecretsHubReader</c> seam. A seam fake proves nothing about the header,
+/// the deadline, or the conversion — all three live only in this class, and all three are exactly
+/// where a follower silently talks to the hub wrong.
+/// </remarks>
+public sealed class GrpcSecretsHubClientTests
+{
+    private static readonly DateTimeOffset Now = new(2026, 8, 7, 12, 0, 0, TimeSpan.Zero);
+
+    private readonly FakeSecretsHubCallInvoker _invoker = new();
+    private readonly RecordingLogger<GrpcSecretsHubClient> _logger = new();
+
+    private readonly SecretsGrpcHubClientOptions _options = new()
+    {
+        Endpoint = "https://central:5001",
+        BearerToken = "hub-token",
+        CallDeadline = TimeSpan.FromSeconds(17),
+    };
+
+    private GrpcSecretsHubClient CreateClient() => new(
+        new SecretsHub.SecretsHubClient(_invoker),
+        Options.Create(_options),
+        _logger,
+        new FixedTimeProvider(Now));
+
+    [Fact]
+    public async Task Every_call_presents_the_bearer_credential()
+    {
+        _invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("db/password") } };
+        _invoker.SecretsResponse = new GetSecretsResponse { Secrets = { TestWire.Row("db/password", 0xAA) } };
+
+        GrpcSecretsHubClient client = CreateClient();
+
+        await client.GetManifestAsync(CancellationToken.None);
+        await client.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
+
+        Assert.Equal(["GetManifest", "GetSecrets"], _invoker.Methods);
+        Assert.Equal("Bearer hub-token", _invoker.AuthorizationOn(0));
+        Assert.Equal("Bearer hub-token", _invoker.AuthorizationOn(1));
+    }
+
+    [Fact]
+    public async Task Every_call_carries_a_deadline_of_now_plus_the_configured_budget()
+    {
+        await CreateClient().GetManifestAsync(CancellationToken.None);
+
+        Assert.Equal(Now.UtcDateTime.AddSeconds(17), _invoker.Calls[0].Deadline);
+    }
+
+    [Fact]
+    public async Task The_deadline_is_recomputed_per_call_rather_than_cached()
+    {
+        // A CallOptions built once at construction would carry an absolute deadline that expires
+        // shortly after startup and then fails every sweep forever, which reads as a hub outage.
+        GrpcSecretsHubClient client = CreateClient();
+
+        await client.GetManifestAsync(CancellationToken.None);
+        await client.GetManifestAsync(CancellationToken.None);
+
+        Assert.Equal(_invoker.Calls[0].Deadline, _invoker.Calls[1].Deadline);
+        Assert.NotSame(_invoker.Calls[0].Headers, _invoker.Calls[1].Headers);
+    }
+
+    [Fact]
+    public async Task The_fetch_asks_for_exactly_the_names_it_was_given()
+    {
+        _invoker.SecretsResponse = new GetSecretsResponse();
+
+        await CreateClient().GetManyAsync(
+            [new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
+
+        Assert.Equal(["db/password", "api/key"], _invoker.LastSecretsRequest!.Names);
+    }
+
+    [Fact]
+    public async Task A_wire_row_comes_back_as_a_domain_row()
+    {
+        _invoker.SecretsResponse = new GetSecretsResponse { Secrets = { TestWire.Row("db/password", 0xAA) } };
+
+        IReadOnlyList<StoredSecret> rows =
+            await CreateClient().GetManyAsync([new SecretName("db/password")], CancellationToken.None);
+
+        StoredSecret row = Assert.Single(rows);
+        Assert.Equal("db/password", row.Name.Value);
+        Assert.Equal(SecretContentType.Text, row.ContentType);
+        Assert.Equal(new byte[] { 0xAA }, row.Ciphertext);
+        Assert.Equal(TestWire.FixedInstant, row.UpdatedUtc);
+    }
+
+    [Fact]
+    public async Task A_wire_manifest_entry_comes_back_as_a_domain_entry()
+    {
+        _invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("api/key") } };
+
+        IReadOnlyList<SecretManifestEntry> entries =
+            await CreateClient().GetManifestAsync(CancellationToken.None);
+
+        SecretManifestEntry entry = Assert.Single(entries);
+        Assert.Equal("api/key", entry.Name.Value);
+        Assert.Equal(TestWire.FixedInstant, entry.UpdatedUtc);
+    }
+
+    [Fact]
+    public async Task A_malformed_row_is_dropped_and_its_well_formed_siblings_survive()
+    {
+        // The whole point: one unparseable row must not cost the batch. Eagerly materializing the
+        // response would fail all three, and the identical pull set would re-fail every interval.
+        SecretRow bad = TestWire.Row("api/key", 0xBB);
+        bad.ContentType = "Kerberos";
+
+        _invoker.SecretsResponse = new GetSecretsResponse
+        {
+            Secrets = { TestWire.Row("db/password", 0xAA), bad, TestWire.Row("svc/token", 0xCC) },
+        };
+
+        IReadOnlyList<StoredSecret> rows = await CreateClient().GetManyAsync(
+            [new SecretName("db/password"), new SecretName("api/key"), new SecretName("svc/token")],
+            CancellationToken.None);
+
+        Assert.Equal(["db/password", "svc/token"], rows.Select(r => r.Name.Value));
+    }
+
+    [Fact]
+    public async Task A_dropped_row_costs_exactly_one_warning_that_names_it()
+    {
+        SecretRow bad = TestWire.Row("api/key", 0xBB);
+        bad.ContentType = "Kerberos";
+
+        _invoker.SecretsResponse = new GetSecretsResponse
+        {
+            Secrets = { TestWire.Row("db/password", 0xAA), bad },
+        };
+
+        await CreateClient().GetManyAsync(
+            [new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
+
+        string warning = Assert.Single(_logger.Warnings);
+        Assert.Contains("api/key", warning, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public async Task A_dropped_row_never_logs_the_material_it_was_carrying()
+    {
+        // A parse failure is an operations problem, so it is logged — but a replicated row is
+        // ciphertext plus its wrapped DEK, and none of that may reach a log sink.
+        SecretRow bad = TestWire.Row("api/key", 0xBB);
+        bad.UpdatedUtc = "yesterday";
+
+        _invoker.SecretsResponse = new GetSecretsResponse { Secrets = { bad } };
+
+        await CreateClient().GetManyAsync([new SecretName("api/key")], CancellationToken.None);
+
+        string warning = Assert.Single(_logger.Warnings);
+
+        foreach (string forbidden in new[] { "sha256:test", "hub-token", "Ciphertext", "WrappedDek" })
+        {
+            Assert.DoesNotContain(forbidden, warning, StringComparison.OrdinalIgnoreCase);
+        }
+    }
+
+    [Fact]
+    public async Task A_row_whose_own_name_is_unparseable_is_dropped_without_echoing_it()
+    {
+        // The name is the one field a warning may legitimately carry — but here the name IS what
+        // failed to parse, so it is replaced by a marker rather than trusted into the log line.
+        SecretRow bad = TestWire.Row("db/password", 0xAA);
+        bad.Name = "../../etc/passwd";
+
+        _invoker.SecretsResponse = new GetSecretsResponse { Secrets = { bad } };
+
+        IReadOnlyList<StoredSecret> rows =
+            await CreateClient().GetManyAsync([new SecretName("db/password")], CancellationToken.None);
+
+        Assert.Empty(rows);
+
+        string warning = Assert.Single(_logger.Warnings);
+        Assert.DoesNotContain("etc/passwd", warning, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public async Task A_malformed_manifest_entry_is_dropped_and_the_rest_of_the_manifest_survives()
+    {
+        ManifestEntry bad = TestWire.Entry("api/key");
+        bad.UpdatedUtc = "not-a-timestamp";
+
+        _invoker.ManifestResponse = new GetManifestResponse
+        {
+            Entries = { TestWire.Entry("db/password"), bad, TestWire.Entry("svc/token") },
+        };
+
+        IReadOnlyList<SecretManifestEntry> entries =
+            await CreateClient().GetManifestAsync(CancellationToken.None);
+
+        Assert.Equal(["db/password", "svc/token"], entries.Select(e => e.Name.Value));
+        Assert.Single(_logger.Warnings);
+    }
+
+    [Fact]
+    public async Task Conversion_preserves_the_order_the_hub_sent()
+    {
+        _invoker.SecretsResponse = new GetSecretsResponse
+        {
+            Secrets = { TestWire.Row("db/password", 0xAA), TestWire.Row("api/key", 0xBB) },
+        };
+
+        IReadOnlyList<StoredSecret> rows = await CreateClient().GetManyAsync(
+            [new SecretName("db/password"), new SecretName("api/key")], CancellationToken.None);
+
+        Assert.Equal(["db/password", "api/key"], rows.Select(r => r.Name.Value));
+    }
+
+    [Fact]
+    public async Task A_well_formed_response_logs_nothing()
+    {
+        _invoker.ManifestResponse = new GetManifestResponse { Entries = { TestWire.Entry("db/password") } };
+
+        await CreateClient().GetManifestAsync(CancellationToken.None);
+
+        Assert.Empty(_logger.Warnings);
+        Assert.Empty(_logger.Information);
+    }
+}
@@ -0,0 +1,186 @@
+using Google.Protobuf;
+using Grpc.Core;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+/// <summary>
+/// A <see cref="CallInvoker"/> that answers the two hub RPCs from canned responses and records what
+/// the client sent.
+/// </summary>
+/// <remarks>
+/// This is what lets the tests drive the <b>real</b> <c>GrpcSecretsHubClient</c> — the generated
+/// client accepts a <see cref="CallInvoker"/>, so substituting one exercises the header, the
+/// deadline and the wire-to-domain conversion without a live server. A fake of the
+/// <c>ISecretsHubReader</c> seam would bypass every one of those.
+/// </remarks>
+internal sealed class FakeSecretsHubCallInvoker : CallInvoker
+{
+    /// <summary>Response returned for <c>GetManifest</c>.</summary>
+    public GetManifestResponse ManifestResponse { get; set; } = new();
+
+    /// <summary>Response returned for <c>GetSecrets</c>.</summary>
+    public GetSecretsResponse SecretsResponse { get; set; } = new();
+
+    /// <summary>The options every call was invoked with, in call order.</summary>
+    public List<CallOptions> Calls { get; } = [];
+
+    /// <summary>The RPC names invoked, in call order.</summary>
+    public List<string> Methods { get; } = [];
+
+    /// <summary>The most recent <c>GetSecrets</c> request, so tests can assert what was asked for.</summary>
+    public GetSecretsRequest? LastSecretsRequest { get; private set; }
+
+    /// <summary>Returns the <c>Authorization</c> header value seen on the call at the given index.</summary>
+    /// <param name="index">Zero-based call index.</param>
+    /// <returns>The header value, or <see langword="null"/> when the header was not sent.</returns>
+    public string? AuthorizationOn(int index) =>
+        Calls[index].Headers?.GetValue("authorization");
+
+    /// <inheritdoc />
+    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
+        Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(method);
+
+        Calls.Add(options);
+        Methods.Add(method.Name);
+
+        if (request is GetSecretsRequest secretsRequest)
+        {
+            LastSecretsRequest = secretsRequest;
+        }
+
+        object response = method.Name switch
+        {
+            "GetManifest" => ManifestResponse,
+            "GetSecrets" => SecretsResponse,
+            _ => throw new InvalidOperationException($"unexpected RPC '{method.Name}'."),
+        };
+
+        return new AsyncUnaryCall<TResponse>(
+            Task.FromResult((TResponse)response),
+            Task.FromResult(new Metadata()),
+            () => Status.DefaultSuccess,
+            () => new Metadata(),
+            () => { });
+    }
+
+    /// <inheritdoc />
+    public override TResponse BlockingUnaryCall<TRequest, TResponse>(
+        Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request) =>
+        throw new NotSupportedException("The hub client only makes async unary calls.");
+
+    /// <inheritdoc />
+    public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(
+        Method<TRequest, TResponse> method, string? host, CallOptions options, TRequest request) =>
+        throw new NotSupportedException("The hub contract has no streaming RPC.");
+
+    /// <inheritdoc />
+    public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(
+        Method<TRequest, TResponse> method, string? host, CallOptions options) =>
+        throw new NotSupportedException("The hub contract has no streaming RPC.");
+
+    /// <inheritdoc />
+    public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(
+        Method<TRequest, TResponse> method, string? host, CallOptions options) =>
+        throw new NotSupportedException("The hub contract has no streaming RPC.");
+}
+
+/// <summary>A clock frozen at a caller-chosen instant, so a deadline assertion can be exact.</summary>
+/// <param name="now">The instant every read returns.</param>
+internal sealed class FixedTimeProvider(DateTimeOffset now) : TimeProvider
+{
+    /// <inheritdoc />
+    public override DateTimeOffset GetUtcNow() => now;
+}
+
+/// <summary>Captures level and formatted message for every log entry.</summary>
+/// <typeparam name="TCategory">The logger category.</typeparam>
+internal sealed class RecordingLogger<TCategory> : ILogger<TCategory>
+{
+    /// <summary>Formatted messages logged at <see cref="LogLevel.Information"/>.</summary>
+    public List<string> Information { get; } = [];
+
+    /// <summary>Formatted messages logged at <see cref="LogLevel.Warning"/>.</summary>
+    public List<string> Warnings { get; } = [];
+
+    /// <summary>Count of information entries.</summary>
+    public int InformationCount => Information.Count;
+
+    /// <summary>Count of warning entries.</summary>
+    public int WarningCount => Warnings.Count;
+
+    /// <inheritdoc />
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+    /// <inheritdoc />
+    public bool IsEnabled(LogLevel logLevel) => true;
+
+    /// <inheritdoc />
+    public void Log<TState>(
+        LogLevel logLevel,
+        EventId eventId,
+        TState state,
+        Exception? exception,
+        Func<TState, Exception?, string> formatter)
+    {
+        ArgumentNullException.ThrowIfNull(formatter);
+
+        string message = formatter(state, exception);
+
+        switch (logLevel)
+        {
+            case LogLevel.Information:
+                Information.Add(message);
+                break;
+            case LogLevel.Warning:
+                Warnings.Add(message);
+                break;
+            default:
+                break;
+        }
+    }
+}
+
+/// <summary>Builders for well-formed wire messages that individual tests then damage.</summary>
+internal static class TestWire
+{
+    /// <summary>A well-formed wire row.</summary>
+    /// <param name="name">The secret name.</param>
+    /// <param name="marker">A byte written into the ciphertext so the row is identifiable.</param>
+    /// <returns>The wire row.</returns>
+    public static SecretRow Row(string name, byte marker) => new()
+    {
+        Name = name,
+        ContentType = nameof(SecretContentType.Text),
+        Ciphertext = ByteString.CopyFrom(marker),
+        Nonce = ByteString.CopyFrom((byte)1),
+        Tag = ByteString.CopyFrom((byte)2),
+        WrappedDek = ByteString.CopyFrom((byte)3),
+        WrapNonce = ByteString.CopyFrom((byte)4),
+        WrapTag = ByteString.CopyFrom((byte)5),
+        KekId = "sha256:test",
+        Revision = 0,
+        IsDeleted = false,
+        CreatedUtc = FixedInstant.ToString("O"),
+        UpdatedUtc = FixedInstant.ToString("O"),
+    };
+
+    /// <summary>A well-formed manifest entry matching <see cref="Row"/>.</summary>
+    /// <param name="name">The secret name.</param>
+    /// <returns>The wire manifest entry.</returns>
+    public static ManifestEntry Entry(string name) => new()
+    {
+        Name = name,
+        Revision = 0,
+        UpdatedUtc = FixedInstant.ToString("O"),
+        IsDeleted = false,
+    };
+
+    /// <summary>The timestamp every builder stamps, so rows differ only where a test makes them differ.</summary>
+    public static readonly DateTimeOffset FixedInstant =
+        new(2026, 8, 7, 1, 2, 3, TimeSpan.Zero);
+}
@@ -1,3 +1,4 @@
+using System.Reflection;
 using Google.Protobuf;
 using ZB.MOM.WW.Secrets.Abstractions;
 using ZB.MOM.WW.Secrets.Replicator.Grpc.V1;
@@ -192,14 +193,18 @@ public sealed class WireToDomainTests
    }

    [Fact]
-    public void Batch_conversions_preserve_order()
+    public void There_is_no_batch_entry_point_to_convert_a_whole_response_with()
    {
-        SecretRow first = FullRow();
-        SecretRow second = FullRow();
-        second.Name = "api/key";
+        // A batch method here would fail a whole response on one bad element, and since the pull set
+        // is derived from the hub's manifest the identical bad element would re-fail the identical
+        // sweep forever. Isolation belongs to GrpcSecretsHubClient, which converts item by item, so
+        // this pins that the strict parser offers nothing that could poison a batch again.
+        string[] members = typeof(WireToDomain)
+            .GetMethods(BindingFlags.Public | BindingFlags.Static | BindingFlags.DeclaredOnly)
+            .Select(m => m.Name)
+            .Order(StringComparer.Ordinal)
+            .ToArray();

-        IReadOnlyList<StoredSecret> rows = WireToDomain.ToRows([first, second]);
-
-        Assert.Equal(["db/password", "api/key"], rows.Select(r => r.Name.Value));
+        Assert.Equal(["ToEntry", "ToRow"], members);
    }
 }