From fac3c5355c3c3f642c67ec08e835c340d2c72a0c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 15:23:02 -0500 Subject: [PATCH] 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 --- .../SecretsGrpcHubClientExtensions.cs | 16 +- .../FailoverSecretsHubReader.cs | 7 + .../GrpcSecretSyncService.cs | 7 + .../HubCancellationTests.cs | 148 ++++++++++++++++++ 4 files changed, 174 insertions(+), 4 deletions(-) create mode 100644 ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/HubCancellationTests.cs diff --git a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/DependencyInjection/SecretsGrpcHubClientExtensions.cs b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/DependencyInjection/SecretsGrpcHubClientExtensions.cs index 9646e05..c6c8508 100644 --- a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/DependencyInjection/SecretsGrpcHubClientExtensions.cs +++ b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/DependencyInjection/SecretsGrpcHubClientExtensions.cs @@ -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>().Value.Endpoint)); + sp.GetRequiredService>().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>() - .Value.FallbackEndpoints[index])); + (sp, _) => GrpcChannel.ForAddress( + sp.GetRequiredService>() + .Value.FallbackEndpoints[index], + ChannelOptions())); } services.TryAddSingleton(sp => @@ -173,4 +175,10 @@ public static class SecretsGrpcHubClientExtensions /// The service key. 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 }; } diff --git a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/FailoverSecretsHubReader.cs b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/FailoverSecretsHubReader.cs index 161fdc8..b893717 100644 --- a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/FailoverSecretsHubReader.cs +++ b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/FailoverSecretsHubReader.cs @@ -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; diff --git a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/GrpcSecretSyncService.cs b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/GrpcSecretSyncService.cs index 0194065..28398ba 100644 --- a/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/GrpcSecretSyncService.cs +++ b/ZB.MOM.WW.Secrets/src/ZB.MOM.WW.Secrets.Replicator.Grpc/GrpcSecretSyncService.cs @@ -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 diff --git a/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/HubCancellationTests.cs b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/HubCancellationTests.cs new file mode 100644 index 0000000..480925a --- /dev/null +++ b/ZB.MOM.WW.Secrets/tests/ZB.MOM.WW.Secrets.Replicator.Grpc.Tests/HubCancellationTests.cs @@ -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; + +/// +/// Secrets-019: cancelling a hub call (host shutdown) must surface as cancellation — not as +/// (Cancelled), which the sweep recorded as a failed sync and the failover +/// reader answered by dialing every other endpoint with the same cancelled token. +/// +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(); + _ = Task.Run(async () => + { + try { while (true) { held.Add(await listener.AcceptTcpClientAsync()); } } + catch (Exception) { /* listener stopped */ } + }); + + IConfiguration configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["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(new NullStore()); + services.AddZbSecretsGrpcHubClient(configuration); + await using ServiceProvider sp = services.BuildServiceProvider(); + ISecretsHubReader hub = sp.GetRequiredService(); + + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300)); + Exception ex = await Record.ExceptionAsync(() => hub.GetManifestAsync(cts.Token)); + + Assert.IsAssignableFrom(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(); + var reader = new FailoverSecretsHubReader([primary, fallback], ["https://a:5001", "https://b:5001"], logger); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync(() => 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()); + + 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(); + 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? fault) : ISecretsHubReader + { + public int Calls; + + public Task> GetManifestAsync(CancellationToken ct) + { + Interlocked.Increment(ref Calls); + return fault is null + ? Task.FromResult>([]) + : Task.FromException>(fault()); + } + + public Task> GetManyAsync(IReadOnlyList names, CancellationToken ct) => + Task.FromResult>([]); + } + + private sealed class NullStore : ISecretStore + { + public Task GetAsync(SecretName name, CancellationToken ct) => Task.FromResult(null); + public Task UpsertAsync(StoredSecret row, CancellationToken ct) => Task.CompletedTask; + public Task DeleteAsync(SecretName name, string? actor, CancellationToken ct) => Task.FromResult(false); + public Task> ListAsync(bool includeDeleted, CancellationToken ct) => + Task.FromResult>([]); + public Task> GetManifestAsync(CancellationToken ct) => + Task.FromResult>([]); + public Task ApplyReplicatedAsync(StoredSecret row, CancellationToken ct) => Task.CompletedTask; + public Task ApplyRewrapAsync(StoredSecret rewrappedRow, byte[] expectedCurrentWrappedDek, CancellationToken ct) => + Task.FromResult(false); + } +}