fix(secrets-grpc): hub-call cancellation surfaces as cancellation, not a failed sweep (scadaproj#45)

Secrets-019: the hub channels are created with ThrowOperationCanceledOnCancellation = true, so a call cancelled through its token throws OperationCanceledException. Defensively, the failover reader rethrows RpcException(Cancelled)-with-a-cancelled-token as cancellation instead of dialing every fallback with the same cancelled token, and the sweep treats it as shutdown, not a failure recorded in ISecretsHubSyncStatus. A server-side Cancelled status (token not cancelled) still fails over. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 15:23:02 -05:00
parent 3aa7c728d5
commit fac3c5355c
4 changed files with 174 additions and 4 deletions
@@ -87,7 +87,8 @@ public static class SecretsGrpcHubClientExtensions
        // would silently lose to that registration and this client would then dial someone else's
        // endpoint with the hub credential attached.
        services.TryAddKeyedSingleton(ChannelServiceKey, (sp, _) => GrpcChannel.ForAddress(
-            sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>().Value.Endpoint));
+            sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>().Value.Endpoint,
+            ChannelOptions()));

        if (eager.FallbackEndpoints.Count == 0)
        {
@@ -112,9 +113,10 @@ public static class SecretsGrpcHubClientExtensions
                int index = i;
                services.TryAddKeyedSingleton(
                    FallbackChannelServiceKey(index),
-                    (sp, _) => GrpcChannel.ForAddress(sp
-                        .GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>()
-                        .Value.FallbackEndpoints[index]));
+                    (sp, _) => GrpcChannel.ForAddress(
+                        sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>()
+                            .Value.FallbackEndpoints[index],
+                        ChannelOptions()));
            }

            services.TryAddSingleton<ISecretsHubReader>(sp =>
@@ -173,4 +175,10 @@ public static class SecretsGrpcHubClientExtensions
    /// <returns>The service key.</returns>
    internal static string FallbackChannelServiceKey(int index) =>
        $"{ChannelServiceKey}:fallback:{index}";
+
+    // Secrets-019: a call cancelled through its token must surface as OperationCanceledException,
+    // which is what the sweep treats as shutdown and the failover reader propagates — not as
+    // RpcException(Cancelled), the gRPC default, which the sweep recorded as a failed sync and the
+    // failover reader answered by dialing every other endpoint with the same cancelled token.
+    internal static GrpcChannelOptions ChannelOptions() => new() { ThrowOperationCanceledOnCancellation = true };
 }
@@ -111,6 +111,13 @@ internal sealed partial class FailoverSecretsHubReader : ISecretsHubReader
                Volatile.Write(ref _stickyIndex, index);
                return result;
            }
+            catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && ct.IsCancellationRequested)
+            {
+                // The caller cancelled (host stopping) and the channel reported it gRPC-style. That
+                // is not "this endpoint did not answer": propagate as cancellation, never fail over
+                // with a token that is already cancelled (Secrets-019).
+                throw new OperationCanceledException(ex.Message, ex, ct);
+            }
            catch (RpcException ex)
            {
                lastFailure = ex;
@@ -1,3 +1,4 @@
+using Grpc.Core;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -177,6 +178,12 @@ internal sealed partial class GrpcSecretSyncService : BackgroundService
        {
            // Host shutdown — not a failure.
        }
+        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && ct.IsCancellationRequested)
+        {
+            // Host shutdown, reported gRPC-style by a channel not created through this library's
+            // registration (which sets ThrowOperationCanceledOnCancellation) — also not a failure
+            // (Secrets-019).
+        }
        catch (Exception ex)
        {
            // Deliberately catch-all. Everything reachable from here — an unreachable hub, a
@@ -0,0 +1,148 @@
+using System.Net;
+using System.Net.Sockets;
+using Grpc.Core;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Secrets.Abstractions;
+using ZB.MOM.WW.Secrets.Replicator.Grpc.DependencyInjection;
+
+namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
+
+/// <summary>
+/// Secrets-019: cancelling a hub call (host shutdown) must surface as cancellation — not as
+/// <see cref="RpcException"/>(Cancelled), which the sweep recorded as a failed sync and the failover
+/// reader answered by dialing every other endpoint with the same cancelled token.
+/// </summary>
+public sealed class HubCancellationTests
+{
+    private static RpcException Cancelled() => new(new Status(StatusCode.Cancelled, "Call canceled by the client."));
+
+    [Fact]
+    public async Task The_registered_channel_reports_a_cancelled_call_as_OperationCanceledException()
+    {
+        // A hub that accepts the connection and never answers: the call can only end by cancellation.
+        using var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        int port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        var held = new List<TcpClient>();
+        _ = Task.Run(async () =>
+        {
+            try { while (true) { held.Add(await listener.AcceptTcpClientAsync()); } }
+            catch (Exception) { /* listener stopped */ }
+        });
+
+        IConfiguration configuration = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>
+            {
+                ["Secrets:GrpcHub:Endpoint"] = $"http://127.0.0.1:{port}",
+                ["Secrets:GrpcHub:BearerToken"] = "token-for-a-silent-hub",
+                ["Secrets:GrpcHub:SyncInterval"] = "01:00:00",
+                ["Secrets:GrpcHub:SyncOnStartup"] = "false",
+            })
+            .Build();
+        var services = new ServiceCollection();
+        services.AddLogging();
+        services.AddSingleton<ISecretStore>(new NullStore());
+        services.AddZbSecretsGrpcHubClient(configuration);
+        await using ServiceProvider sp = services.BuildServiceProvider();
+        ISecretsHubReader hub = sp.GetRequiredService<ISecretsHubReader>();
+
+        using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300));
+        Exception ex = await Record.ExceptionAsync(() => hub.GetManifestAsync(cts.Token));
+
+        Assert.IsAssignableFrom<OperationCanceledException>(ex);
+        listener.Stop();
+        held.ForEach(c => c.Dispose());
+    }
+
+    [Fact]
+    public async Task The_failover_reader_does_not_rotate_a_cancelled_call_through_the_fallbacks()
+    {
+        var primary = new FaultingReader(Cancelled);
+        var fallback = new FaultingReader(Cancelled);
+        var logger = new RecordingLogger<FailoverSecretsHubReader>();
+        var reader = new FailoverSecretsHubReader([primary, fallback], ["https://a:5001", "https://b:5001"], logger);
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => reader.GetManifestAsync(cts.Token));
+
+        Assert.Equal(1, primary.Calls);
+        Assert.Equal(0, fallback.Calls);
+        Assert.Equal(0, logger.WarningCount);
+    }
+
+    [Fact]
+    public async Task A_cancelled_status_without_a_cancelled_token_still_fails_over()
+    {
+        // Positive control: Cancelled from the SERVER (token not cancelled) is an endpoint failure.
+        var primary = new FaultingReader(Cancelled);
+        var fallback = new FaultingReader(fault: null);
+        var reader = new FailoverSecretsHubReader(
+            [primary, fallback], ["https://a:5001", "https://b:5001"], new RecordingLogger<FailoverSecretsHubReader>());
+
+        await reader.GetManifestAsync(CancellationToken.None);
+
+        Assert.Equal(1, fallback.Calls);
+    }
+
+    [Fact]
+    public async Task A_sweep_cancelled_gRPC_style_at_shutdown_is_not_recorded_as_a_failure()
+    {
+        var status = new SecretsHubSyncStatus();
+        var logger = new RecordingLogger<GrpcSecretSyncService>();
+        var svc = new GrpcSecretSyncService(
+            new NullStore(),
+            new FaultingReader(Cancelled),
+            cacheInvalidator: null,
+            Options.Create(new SecretsGrpcHubClientOptions
+            {
+                Endpoint = "https://central:5001",
+                BearerToken = "unused",
+                SyncInterval = TimeSpan.FromHours(1),
+            }),
+            logger,
+            TimeProvider.System,
+            status);
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+
+        await svc.SweepForTestAsync(cts.Token);
+
+        Assert.Equal(0, status.ConsecutiveFailures);
+        Assert.Null(status.LastError);
+        Assert.Equal(0, logger.WarningCount);
+    }
+
+    private sealed class FaultingReader(Func<Exception>? fault) : ISecretsHubReader
+    {
+        public int Calls;
+
+        public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct)
+        {
+            Interlocked.Increment(ref Calls);
+            return fault is null
+                ? Task.FromResult<IReadOnlyList<SecretManifestEntry>>([])
+                : Task.FromException<IReadOnlyList<SecretManifestEntry>>(fault());
+        }
+
+        public Task<IReadOnlyList<StoredSecret>> GetManyAsync(IReadOnlyList<SecretName> names, CancellationToken ct) =>
+            Task.FromResult<IReadOnlyList<StoredSecret>>([]);
+    }
+
+    private sealed class NullStore : ISecretStore
+    {
+        public Task<StoredSecret?> GetAsync(SecretName name, CancellationToken ct) => Task.FromResult<StoredSecret?>(null);
+        public Task UpsertAsync(StoredSecret row, CancellationToken ct) => Task.CompletedTask;
+        public Task<bool> DeleteAsync(SecretName name, string? actor, CancellationToken ct) => Task.FromResult(false);
+        public Task<IReadOnlyList<SecretMetadata>> ListAsync(bool includeDeleted, CancellationToken ct) =>
+            Task.FromResult<IReadOnlyList<SecretMetadata>>([]);
+        public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct) =>
+            Task.FromResult<IReadOnlyList<SecretManifestEntry>>([]);
+        public Task ApplyReplicatedAsync(StoredSecret row, CancellationToken ct) => Task.CompletedTask;
+        public Task<bool> ApplyRewrapAsync(StoredSecret rewrappedRow, byte[] expectedCurrentWrappedDek, CancellationToken ct) =>
+            Task.FromResult(false);
+    }
+}