feat(secrets-grpc): FallbackEndpoints + sticky failover hub reader (scadaproj#4)
SecretsGrpcHubClientOptions gains a getter-only FallbackEndpoints list:
additional hub addresses tried in order when a call fails with a
transport-level RpcException. Validation fails startup on blank/relative
entries and on any duplicate of the primary or another fallback (a
duplicated endpoint silently halves the configured failover).
New FailoverSecretsHubReader composes one GrpcSecretsHubClient per
endpoint: the answering endpoint is sticky (a sweep is GetManifest + N
GetSecrets batches — without stickiness every call after a primary
outage re-pays a full CallDeadline before reaching the healthy
fallback), rotation wraps so a recovered primary is retried when the
stuck-to endpoint fails, only RpcException triggers failover
(cancellation and defects propagate untouched), and when every endpoint
fails the last RpcException is rethrown so the sweep degrades to
last-known-good exactly as today. Failover warnings name only the two
endpoint URIs and the status code — never any credential material.
DI: zero fallbacks keeps the registration graph byte-identical to 0.4.x
(single keyed channel under ChannelServiceKey, plain
GrpcSecretsHubClient); with fallbacks, one container-owned keyed channel
per endpoint (fallback i under "{ChannelServiceKey}:fallback:{i}") wired
into the failover reader. GrpcSecretsHubClient itself is unchanged.
Safe only when every listed endpoint serves the SAME store — for
ScadaBridge, the central pair over one shared SQL Server store; the
options doc and CHANGELOG carry the warning.
Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
This commit is contained in:
@@ -5,6 +5,34 @@ follows [Keep a Changelog](https://keepachangelog.com/); versions are shared acr
|
||||
package in the family (see `Directory.Build.props`) even when a given release only touches
|
||||
one of them.
|
||||
|
||||
## 0.5.0 — Unreleased
|
||||
|
||||
### Added
|
||||
|
||||
- **`ZB.MOM.WW.Secrets.Replicator.Grpc` — hub endpoint failover for followers**
|
||||
(scadaproj#4). `SecretsGrpcHubClientOptions` gains `FallbackEndpoints`: additional hub
|
||||
addresses tried, in order, when a call to the primary `Endpoint` (or an earlier fallback)
|
||||
fails with a transport-level gRPC error. The answering endpoint is sticky — a sweep that
|
||||
failed over does not re-pay the dead endpoint's full `CallDeadline` on every subsequent
|
||||
call — and rotation wraps, so a recovered primary is retried the next time the stuck-to
|
||||
endpoint fails. Only `RpcException` triggers failover; cancellation and everything else
|
||||
propagate untouched, and when every endpoint fails the last `RpcException` is rethrown so
|
||||
the sweep degrades to last-known-good exactly as before. Each failover logs a Warning
|
||||
naming the failed and next endpoint URIs (never any credential material).
|
||||
- **Same-store requirement:** list only endpoints that serve the SAME store — for
|
||||
ScadaBridge, the central pair over one shared SQL Server store. Endpoints backed by
|
||||
independent stores turn a primary outage into a silent convergence stop (failing over
|
||||
to an emptier hub "succeeds"), which is the exact defect scadaproj#4 recorded.
|
||||
- Validation fails startup on a blank/relative fallback or one duplicating the primary
|
||||
or another fallback (a duplicated endpoint silently halves the configured failover).
|
||||
|
||||
### Unchanged
|
||||
|
||||
- With `FallbackEndpoints` empty (the default), the registration graph is identical to
|
||||
0.4.x — one keyed channel, one `GrpcSecretsHubClient`, no failover wrapper — and the wire
|
||||
behavior of every call is byte-identical. Every other package in the family is
|
||||
behaviorally unchanged in this release.
|
||||
|
||||
## 0.4.1 — 2026-08-07
|
||||
|
||||
### Fixed
|
||||
|
||||
+77
-6
@@ -45,6 +45,13 @@ public static class SecretsGrpcHubClientExtensions
|
||||
/// means local writes never leave the node, so there is nothing to publish and the store stays
|
||||
/// exactly as the application configured it. A host that forgets <c>AddZbSecrets</c> fails loudly
|
||||
/// at startup when the sweep cannot resolve the store, rather than silently replicating nothing.
|
||||
/// <para>
|
||||
/// When <see cref="SecretsGrpcHubClientOptions.FallbackEndpoints"/> is non-empty, one keyed
|
||||
/// channel and one client are registered per endpoint and the reader becomes a
|
||||
/// <see cref="FailoverSecretsHubReader"/> composed over them, primary first. With the list
|
||||
/// empty — the default — the registration graph is identical to releases before the list
|
||||
/// existed: one keyed channel, one <see cref="GrpcSecretsHubClient"/>, no failover wrapper.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="services">The service collection to add to.</param>
|
||||
/// <param name="configuration">The application configuration.</param>
|
||||
@@ -82,15 +89,79 @@ public static class SecretsGrpcHubClientExtensions
|
||||
services.TryAddKeyedSingleton(ChannelServiceKey, (sp, _) => GrpcChannel.ForAddress(
|
||||
sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>().Value.Endpoint));
|
||||
|
||||
services.TryAddSingleton<ISecretsHubReader>(sp => new GrpcSecretsHubClient(
|
||||
new SecretsHub.SecretsHubClient(
|
||||
sp.GetRequiredKeyedService<GrpcChannel>(ChannelServiceKey)),
|
||||
sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>(),
|
||||
sp.GetRequiredService<ILogger<GrpcSecretsHubClient>>(),
|
||||
sp.GetService<TimeProvider>()));
|
||||
if (eager.FallbackEndpoints.Count == 0)
|
||||
{
|
||||
// No fallbacks: the graph is exactly the pre-0.5.0 one — the single keyed channel above
|
||||
// plus one GrpcSecretsHubClient as the reader. Kept as its own branch rather than a
|
||||
// one-element failover composition so existing consumers see an identical container
|
||||
// shape and zero added indirection on every call.
|
||||
services.TryAddSingleton<ISecretsHubReader>(sp => new GrpcSecretsHubClient(
|
||||
new SecretsHub.SecretsHubClient(
|
||||
sp.GetRequiredKeyedService<GrpcChannel>(ChannelServiceKey)),
|
||||
sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>(),
|
||||
sp.GetRequiredService<ILogger<GrpcSecretsHubClient>>(),
|
||||
sp.GetService<TimeProvider>()));
|
||||
}
|
||||
else
|
||||
{
|
||||
// One keyed channel per fallback, each a container-owned singleton for the same
|
||||
// disposal reason as the primary. The loop variable is copied because the factory
|
||||
// closes over it and runs long after the loop has finished.
|
||||
for (int i = 0; i < eager.FallbackEndpoints.Count; i++)
|
||||
{
|
||||
int index = i;
|
||||
services.TryAddKeyedSingleton(
|
||||
FallbackChannelServiceKey(index),
|
||||
(sp, _) => GrpcChannel.ForAddress(sp
|
||||
.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>()
|
||||
.Value.FallbackEndpoints[index]));
|
||||
}
|
||||
|
||||
services.TryAddSingleton<ISecretsHubReader>(sp =>
|
||||
{
|
||||
IOptions<SecretsGrpcHubClientOptions> options =
|
||||
sp.GetRequiredService<IOptions<SecretsGrpcHubClientOptions>>();
|
||||
ILogger<GrpcSecretsHubClient> clientLogger =
|
||||
sp.GetRequiredService<ILogger<GrpcSecretsHubClient>>();
|
||||
TimeProvider? timeProvider = sp.GetService<TimeProvider>();
|
||||
|
||||
GrpcSecretsHubClient ReaderOver(object channelKey) => new(
|
||||
new SecretsHub.SecretsHubClient(
|
||||
sp.GetRequiredKeyedService<GrpcChannel>(channelKey)),
|
||||
options,
|
||||
clientLogger,
|
||||
timeProvider);
|
||||
|
||||
IList<string> fallbacks = options.Value.FallbackEndpoints;
|
||||
var readers = new List<ISecretsHubReader>(1 + fallbacks.Count)
|
||||
{
|
||||
ReaderOver(ChannelServiceKey),
|
||||
};
|
||||
var labels = new List<string>(1 + fallbacks.Count) { options.Value.Endpoint };
|
||||
|
||||
for (int index = 0; index < fallbacks.Count; index++)
|
||||
{
|
||||
readers.Add(ReaderOver(FallbackChannelServiceKey(index)));
|
||||
labels.Add(fallbacks[index]);
|
||||
}
|
||||
|
||||
return new FailoverSecretsHubReader(
|
||||
readers, labels, sp.GetRequiredService<ILogger<FailoverSecretsHubReader>>());
|
||||
});
|
||||
}
|
||||
|
||||
services.AddHostedService<GrpcSecretSyncService>();
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Service key for the keyed <see cref="GrpcChannel"/> dialing the fallback endpoint at
|
||||
/// <paramref name="index"/> in <see cref="SecretsGrpcHubClientOptions.FallbackEndpoints"/>.
|
||||
/// The primary endpoint's channel keeps <see cref="ChannelServiceKey"/> unchanged.
|
||||
/// </summary>
|
||||
/// <param name="index">Zero-based index into the fallback list.</param>
|
||||
/// <returns>The service key.</returns>
|
||||
internal static string FallbackChannelServiceKey(int index) =>
|
||||
$"{ChannelServiceKey}:fallback:{index}";
|
||||
}
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.Secrets.Replicator.Grpc;
|
||||
|
||||
/// <summary>
|
||||
/// An <see cref="ISecretsHubReader"/> over several single-endpoint readers: tries each endpoint in
|
||||
/// rotation order until one answers, remembering the one that did.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Safe only when every endpoint serves the same store.</b> This type exists for ScadaBridge's
|
||||
/// central pair over one shared SQL Server store, where either central answers with the truth.
|
||||
/// Composed over endpoints with independent stores it would turn a primary outage into a silent
|
||||
/// convergence stop — the fallback "succeeds" against an emptier inventory (scadaproj#4). The DI
|
||||
/// extension's options doc carries the same warning where operators author the list.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The answering endpoint is sticky.</b> A sweep is one <c>GetManifest</c> plus N
|
||||
/// <c>GetSecrets</c> batches; without stickiness every call after a primary outage would re-pay the
|
||||
/// primary's failure — a full <c>CallDeadline</c> of latency — before reaching the healthy fallback.
|
||||
/// So each call starts at the last endpoint that answered (initially the primary) and rotates from
|
||||
/// there, which also means a recovered primary is retried naturally the next time the stuck-to
|
||||
/// endpoint fails, rather than being probed on a timer.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Only <see cref="RpcException"/> triggers failover.</b> It is the one signal that means "this
|
||||
/// endpoint did not usefully answer". <see cref="OperationCanceledException"/> is the host stopping
|
||||
/// — dialing another endpoint with a cancelled token would only delay shutdown — and anything else
|
||||
/// is a defect, which retrying elsewhere would mask. Both propagate untouched. When every endpoint
|
||||
/// fails, the <em>last</em> <see cref="RpcException"/> is rethrown so the sweep's catch-all handles
|
||||
/// it exactly as it handles a single-endpoint outage today.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed partial class FailoverSecretsHubReader : ISecretsHubReader
|
||||
{
|
||||
private readonly IReadOnlyList<ISecretsHubReader> _readers;
|
||||
private readonly IReadOnlyList<string> _endpointLabels;
|
||||
private readonly ILogger<FailoverSecretsHubReader> _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Index of the last endpoint that answered; every call starts its rotation here. Read and
|
||||
/// written with <see cref="Volatile"/> — a torn or stale value costs at worst one extra failed
|
||||
/// dial, never a wrong result, so a lock would buy nothing.
|
||||
/// </summary>
|
||||
private int _stickyIndex;
|
||||
|
||||
/// <summary>Creates the failover reader.</summary>
|
||||
/// <param name="readers">One reader per endpoint, primary first — rotation order.</param>
|
||||
/// <param name="endpointLabels">
|
||||
/// The endpoint URI each reader dials, index-aligned with <paramref name="readers"/>. Used only
|
||||
/// in log messages: a URI is not a secret, and naming both sides of a failover is what makes the
|
||||
/// warning actionable.
|
||||
/// </param>
|
||||
/// <param name="logger">Logger for failover warnings.</param>
|
||||
/// <exception cref="ArgumentException">
|
||||
/// <paramref name="readers"/> is empty or the two lists differ in length.
|
||||
/// </exception>
|
||||
public FailoverSecretsHubReader(
|
||||
IReadOnlyList<ISecretsHubReader> readers,
|
||||
IReadOnlyList<string> endpointLabels,
|
||||
ILogger<FailoverSecretsHubReader> logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(readers);
|
||||
ArgumentNullException.ThrowIfNull(endpointLabels);
|
||||
|
||||
if (readers.Count == 0)
|
||||
{
|
||||
throw new ArgumentException("At least one hub reader is required.", nameof(readers));
|
||||
}
|
||||
|
||||
if (readers.Count != endpointLabels.Count)
|
||||
{
|
||||
throw new ArgumentException(
|
||||
$"Endpoint labels ({endpointLabels.Count}) must align one-to-one with readers " +
|
||||
$"({readers.Count}).",
|
||||
nameof(endpointLabels));
|
||||
}
|
||||
|
||||
_readers = readers;
|
||||
_endpointLabels = endpointLabels;
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct) =>
|
||||
ExecuteWithFailoverAsync(static (reader, token) => reader.GetManifestAsync(token), ct);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<StoredSecret>> GetManyAsync(
|
||||
IReadOnlyList<SecretName> names, CancellationToken ct) =>
|
||||
ExecuteWithFailoverAsync((reader, token) => reader.GetManyAsync(names, token), ct);
|
||||
|
||||
// The one rotation, shared by both RPCs so their failover behaviour cannot drift apart.
|
||||
private async Task<T> ExecuteWithFailoverAsync<T>(
|
||||
Func<ISecretsHubReader, CancellationToken, Task<T>> call, CancellationToken ct)
|
||||
{
|
||||
int start = Volatile.Read(ref _stickyIndex);
|
||||
RpcException? lastFailure = null;
|
||||
|
||||
for (int attempt = 0; attempt < _readers.Count; attempt++)
|
||||
{
|
||||
int index = (start + attempt) % _readers.Count;
|
||||
|
||||
try
|
||||
{
|
||||
T result = await call(_readers[index], ct).ConfigureAwait(false);
|
||||
Volatile.Write(ref _stickyIndex, index);
|
||||
return result;
|
||||
}
|
||||
catch (RpcException ex)
|
||||
{
|
||||
lastFailure = ex;
|
||||
|
||||
if (attempt < _readers.Count - 1)
|
||||
{
|
||||
// The message carries the two endpoint URIs and the status code — nothing from
|
||||
// the exception detail, which is server-authored text no credential belongs in
|
||||
// but this node cannot vouch for.
|
||||
LogEndpointFailedOver(
|
||||
ex,
|
||||
_endpointLabels[index],
|
||||
ex.StatusCode,
|
||||
_endpointLabels[(start + attempt + 1) % _readers.Count]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Every endpoint failed. Rethrowing the LAST failure hands the sweep's catch-all the
|
||||
// freshest evidence; it logs and retries next tick exactly as for a single endpoint.
|
||||
throw lastFailure!;
|
||||
}
|
||||
|
||||
[LoggerMessage(
|
||||
EventId = 1,
|
||||
Level = LogLevel.Warning,
|
||||
Message = "Secrets hub endpoint {FailedEndpoint} did not answer ({StatusCode}); failing over to {NextEndpoint}.")]
|
||||
private partial void LogEndpointFailedOver(
|
||||
Exception exception, string failedEndpoint, StatusCode statusCode, string nextEndpoint);
|
||||
}
|
||||
+47
-2
@@ -28,6 +28,21 @@ public sealed class SecretsGrpcHubClientOptions
|
||||
/// </remarks>
|
||||
public string Endpoint { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>
|
||||
/// Additional hub addresses tried, in order, when a call to <see cref="Endpoint"/> (or an
|
||||
/// earlier fallback) fails with a transport-level gRPC error. Optional; empty by default.
|
||||
/// Meaningful only when every listed endpoint serves the SAME store — for ScadaBridge that
|
||||
/// is the central pair over one shared SQL Server store. Do NOT list endpoints backed by
|
||||
/// independent stores: failing over to an emptier hub is a silent convergence stop, which is
|
||||
/// the exact defect (scadaproj#4) the shared store exists to prevent.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Getter-only so the configuration binder adds entries to this instance rather than needing a
|
||||
/// setter. Each entry carries the same http-cleartext caveat as <see cref="Endpoint"/>: a plain
|
||||
/// <c>http</c> fallback sends <see cref="BearerToken"/> in cleartext on every failed-over call.
|
||||
/// </remarks>
|
||||
public IList<string> FallbackEndpoints { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Shared bearer token presented on every call. Required — an unset token is a hard startup
|
||||
/// failure rather than an anonymous call, because the hub authenticates on this alone.
|
||||
@@ -69,8 +84,9 @@ public sealed class SecretsGrpcHubClientOptions
|
||||
/// operator actually has to edit rather than the default one.
|
||||
/// </param>
|
||||
/// <exception cref="InvalidOperationException">
|
||||
/// The endpoint is missing or is not an absolute http/https URI, the bearer token is missing, or
|
||||
/// an interval is not positive.
|
||||
/// The endpoint is missing or is not an absolute http/https URI, a fallback endpoint is blank,
|
||||
/// not an absolute http/https URI, or duplicates another endpoint, the bearer token is missing,
|
||||
/// or an interval is not positive.
|
||||
/// </exception>
|
||||
public void Validate(string sectionPath = DefaultSectionPath)
|
||||
{
|
||||
@@ -92,6 +108,35 @@ public sealed class SecretsGrpcHubClientOptions
|
||||
$"({sectionPath}:Endpoint, was '{Endpoint}').");
|
||||
}
|
||||
|
||||
var seenEndpoints = new HashSet<string>(StringComparer.OrdinalIgnoreCase) { Endpoint };
|
||||
|
||||
foreach (string fallback in FallbackEndpoints)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(fallback))
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"gRPC secrets hub client fallback endpoints must not be blank " +
|
||||
$"({sectionPath}:FallbackEndpoints).");
|
||||
}
|
||||
|
||||
if (!Uri.TryCreate(fallback, UriKind.Absolute, out Uri? fallbackUri) ||
|
||||
(fallbackUri.Scheme != Uri.UriSchemeHttp && fallbackUri.Scheme != Uri.UriSchemeHttps))
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"gRPC secrets hub client fallback endpoints must be absolute http or https " +
|
||||
$"URIs ({sectionPath}:FallbackEndpoints, entry '{fallback}').");
|
||||
}
|
||||
|
||||
if (!seenEndpoints.Add(fallback))
|
||||
{
|
||||
// A duplicated endpoint silently halves the failover the operator thinks they
|
||||
// configured: the rotation would try the same address twice and call it two tries.
|
||||
throw new InvalidOperationException(
|
||||
$"gRPC secrets hub client fallback endpoints must not repeat the Endpoint or " +
|
||||
$"each other ({sectionPath}:FallbackEndpoints, entry '{fallback}').");
|
||||
}
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(BearerToken))
|
||||
{
|
||||
// Fail closed. An empty token would otherwise be presented as `Bearer `, which the hub
|
||||
|
||||
+266
@@ -0,0 +1,266 @@
|
||||
using Grpc.Core;
|
||||
using ZB.MOM.WW.Secrets.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Rotation, stickiness and containment behaviour of the multi-endpoint hub reader. Exercised
|
||||
/// against scripted <see cref="ISecretsHubReader"/> fakes: the per-endpoint wire behaviour (headers,
|
||||
/// deadlines, parse isolation) belongs to <c>GrpcSecretsHubClient</c> and is tested there — this
|
||||
/// suite is only about which endpoint gets asked, in what order, and what escapes when they all fail.
|
||||
/// </summary>
|
||||
public sealed class FailoverSecretsHubReaderTests
|
||||
{
|
||||
private const string PrimaryEndpoint = "https://central-a:5001";
|
||||
private const string FallbackEndpoint = "https://central-b:5001";
|
||||
|
||||
private readonly ScriptedHubReader _primary = new();
|
||||
private readonly ScriptedHubReader _fallback = new();
|
||||
private readonly RecordingLogger<FailoverSecretsHubReader> _logger = new();
|
||||
|
||||
private FailoverSecretsHubReader CreateReader() => new(
|
||||
[_primary, _fallback],
|
||||
[PrimaryEndpoint, FallbackEndpoint],
|
||||
_logger);
|
||||
|
||||
private static RpcException Unavailable() => new(new Status(
|
||||
StatusCode.Unavailable, "connection refused"));
|
||||
|
||||
[Fact]
|
||||
public async Task A_healthy_primary_is_the_only_endpoint_asked()
|
||||
{
|
||||
FailoverSecretsHubReader reader = CreateReader();
|
||||
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
await reader.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
|
||||
|
||||
Assert.Equal(1, _primary.ManifestCalls);
|
||||
Assert.Equal(1, _primary.ManyCalls);
|
||||
Assert.Equal(0, _fallback.ManifestCalls);
|
||||
Assert.Equal(0, _fallback.ManyCalls);
|
||||
Assert.Equal(0, _logger.WarningCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_transport_fault_on_the_primary_fails_over_to_the_fallback()
|
||||
{
|
||||
_primary.FaultWith = Unavailable;
|
||||
_fallback.Manifest = [Entry("db/password")];
|
||||
|
||||
IReadOnlyList<SecretManifestEntry> manifest =
|
||||
await CreateReader().GetManifestAsync(CancellationToken.None);
|
||||
|
||||
Assert.Single(manifest);
|
||||
Assert.Equal(1, _primary.ManifestCalls);
|
||||
Assert.Equal(1, _fallback.ManifestCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_failover_warning_names_both_endpoints()
|
||||
{
|
||||
_primary.FaultWith = Unavailable;
|
||||
|
||||
await CreateReader().GetManifestAsync(CancellationToken.None);
|
||||
|
||||
string warning = Assert.Single(_logger.Warnings);
|
||||
Assert.Contains(PrimaryEndpoint, warning, StringComparison.Ordinal);
|
||||
Assert.Contains(FallbackEndpoint, warning, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetMany_fails_over_too()
|
||||
{
|
||||
_primary.FaultWith = Unavailable;
|
||||
_fallback.Rows = [Row("db/password")];
|
||||
|
||||
IReadOnlyList<StoredSecret> rows = await CreateReader()
|
||||
.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
|
||||
|
||||
Assert.Single(rows);
|
||||
Assert.Equal(1, _primary.ManyCalls);
|
||||
Assert.Equal(1, _fallback.ManyCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_endpoint_that_answered_is_sticky_for_the_next_call()
|
||||
{
|
||||
// A sweep is one GetManifest plus N GetSecrets batches. Without stickiness, every call
|
||||
// after a primary outage would re-pay the primary's full CallDeadline of latency before
|
||||
// reaching the fallback that is actually answering.
|
||||
_primary.FaultWith = Unavailable;
|
||||
FailoverSecretsHubReader reader = CreateReader();
|
||||
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
await reader.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
|
||||
|
||||
Assert.Equal(1, _primary.ManifestCalls);
|
||||
Assert.Equal(0, _primary.ManyCalls);
|
||||
Assert.Equal(1, _fallback.ManyCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Rotation_wraps_back_to_a_recovered_primary_when_the_fallback_fails_in_turn()
|
||||
{
|
||||
_primary.FaultWith = Unavailable;
|
||||
FailoverSecretsHubReader reader = CreateReader();
|
||||
|
||||
// Sticks to the fallback…
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
|
||||
// …then the fallback dies and the primary has recovered.
|
||||
_primary.FaultWith = null;
|
||||
_fallback.FaultWith = Unavailable;
|
||||
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(2, _primary.ManifestCalls);
|
||||
Assert.Equal(2, _fallback.ManifestCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task When_every_endpoint_fails_the_last_exception_is_rethrown()
|
||||
{
|
||||
var primaryFault = new RpcException(new Status(StatusCode.Unavailable, "primary down"));
|
||||
var fallbackFault = new RpcException(new Status(StatusCode.DeadlineExceeded, "fallback down"));
|
||||
_primary.FaultWith = () => primaryFault;
|
||||
_fallback.FaultWith = () => fallbackFault;
|
||||
|
||||
RpcException thrown = await Assert.ThrowsAsync<RpcException>(
|
||||
() => CreateReader().GetManifestAsync(CancellationToken.None));
|
||||
|
||||
// The LAST endpoint tried, so the sweep's catch-all logs the freshest failure rather than
|
||||
// one from an endpoint a later attempt already moved past.
|
||||
Assert.Same(fallbackFault, thrown);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Cancellation_propagates_immediately_without_failing_over()
|
||||
{
|
||||
// Cancellation is the host stopping, not an endpoint fault — dialing the fallback with an
|
||||
// already-cancelled token would only delay shutdown by another deadline.
|
||||
_primary.FaultWith = () => new OperationCanceledException();
|
||||
|
||||
await Assert.ThrowsAsync<OperationCanceledException>(
|
||||
() => CreateReader().GetManifestAsync(CancellationToken.None));
|
||||
|
||||
Assert.Equal(0, _fallback.ManifestCalls);
|
||||
Assert.Equal(0, _logger.WarningCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_non_transport_exception_propagates_immediately_without_failing_over()
|
||||
{
|
||||
_primary.FaultWith = () => new InvalidOperationException("a defect, not an outage");
|
||||
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => CreateReader().GetManifestAsync(CancellationToken.None));
|
||||
|
||||
Assert.Equal(0, _fallback.ManifestCalls);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task No_failover_log_message_carries_token_material()
|
||||
{
|
||||
// The warning's whole content is the two endpoint URIs (not secrets) and the gRPC status
|
||||
// code. Even a status detail that happens to quote a credential must not reach the message.
|
||||
const string leak = "s3cr3t-bearer-token";
|
||||
_primary.FaultWith = () => new RpcException(new Status(
|
||||
StatusCode.Unauthenticated, $"Bearer {leak} rejected"));
|
||||
|
||||
await CreateReader().GetManifestAsync(CancellationToken.None);
|
||||
|
||||
foreach (string warning in _logger.Warnings)
|
||||
{
|
||||
Assert.DoesNotContain(leak, warning, StringComparison.OrdinalIgnoreCase);
|
||||
}
|
||||
|
||||
Assert.Equal(1, _logger.WarningCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Concurrent_calls_share_the_sticky_index_without_tearing()
|
||||
{
|
||||
// The sticky index is plain shared state written from every call path; this pins that
|
||||
// parallel sweeps (manifest + batches overlapping) never corrupt it or deadlock.
|
||||
_primary.FaultWith = Unavailable;
|
||||
FailoverSecretsHubReader reader = CreateReader();
|
||||
|
||||
await Task.WhenAll(Enumerable.Range(0, 32).Select(async _ =>
|
||||
{
|
||||
IReadOnlyList<SecretManifestEntry> manifest =
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
Assert.NotNull(manifest);
|
||||
}));
|
||||
|
||||
// Every one of the 32 calls ended on the answering fallback.
|
||||
Assert.Equal(32, _fallback.ManifestCalls);
|
||||
// And the index is now stuck there: one more call never touches the primary.
|
||||
int primaryCallsBefore = _primary.ManifestCalls;
|
||||
await reader.GetManifestAsync(CancellationToken.None);
|
||||
Assert.Equal(primaryCallsBefore, _primary.ManifestCalls);
|
||||
}
|
||||
|
||||
private static SecretManifestEntry Entry(string name) => new()
|
||||
{
|
||||
Name = new SecretName(name),
|
||||
Revision = 0,
|
||||
UpdatedUtc = TestWire.FixedInstant,
|
||||
};
|
||||
|
||||
private static StoredSecret Row(string name) => new()
|
||||
{
|
||||
Name = new SecretName(name),
|
||||
ContentType = SecretContentType.Text,
|
||||
Ciphertext = [0xAA],
|
||||
Nonce = [1],
|
||||
Tag = [2],
|
||||
WrappedDek = [3],
|
||||
WrapNonce = [4],
|
||||
WrapTag = [5],
|
||||
KekId = "sha256:test",
|
||||
Revision = 0,
|
||||
CreatedUtc = TestWire.FixedInstant,
|
||||
UpdatedUtc = TestWire.FixedInstant,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// An <see cref="ISecretsHubReader"/> standing in for one endpoint's client: counts calls,
|
||||
/// answers from canned data, or faults with whatever a test scripts.
|
||||
/// </summary>
|
||||
private sealed class ScriptedHubReader : ISecretsHubReader
|
||||
{
|
||||
/// <summary>Number of <c>GetManifest</c> calls received.</summary>
|
||||
public int ManifestCalls;
|
||||
|
||||
/// <summary>Number of <c>GetMany</c> calls received.</summary>
|
||||
public int ManyCalls;
|
||||
|
||||
/// <summary>When set, every call throws this instead of answering.</summary>
|
||||
public Func<Exception>? FaultWith { get; set; }
|
||||
|
||||
/// <summary>Canned manifest returned when not faulted.</summary>
|
||||
public IReadOnlyList<SecretManifestEntry> Manifest { get; set; } = [];
|
||||
|
||||
/// <summary>Canned rows returned when not faulted.</summary>
|
||||
public IReadOnlyList<StoredSecret> Rows { get; set; } = [];
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<SecretManifestEntry>> GetManifestAsync(CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref ManifestCalls);
|
||||
return FaultWith is { } fault
|
||||
? Task.FromException<IReadOnlyList<SecretManifestEntry>>(fault())
|
||||
: Task.FromResult(Manifest);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<StoredSecret>> GetManyAsync(
|
||||
IReadOnlyList<SecretName> names, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref ManyCalls);
|
||||
return FaultWith is { } fault
|
||||
? Task.FromException<IReadOnlyList<StoredSecret>>(fault())
|
||||
: Task.FromResult(Rows);
|
||||
}
|
||||
}
|
||||
}
|
||||
+76
@@ -452,6 +452,82 @@ public sealed class GrpcSecretSyncServiceTests : IDisposable
|
||||
SecretsGrpcHubClientExtensions.ChannelServiceKey));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Registration_with_no_fallbacks_resolves_the_plain_single_endpoint_client()
|
||||
{
|
||||
// The zero-fallback registration graph is the pre-0.5.0 one, unchanged: one keyed channel,
|
||||
// one GrpcSecretsHubClient as the reader. Consumers wired before FallbackEndpoints existed
|
||||
// must see an identical container shape.
|
||||
ServiceCollection services = new();
|
||||
services.AddLogging();
|
||||
services.AddZbSecretsGrpcHubClient(HubConfig());
|
||||
|
||||
using ServiceProvider provider = services.BuildServiceProvider();
|
||||
|
||||
Assert.IsType<GrpcSecretsHubClient>(provider.GetRequiredService<ISecretsHubReader>());
|
||||
Assert.NotNull(provider.GetRequiredKeyedService<GrpcChannel>(
|
||||
SecretsGrpcHubClientExtensions.ChannelServiceKey));
|
||||
Assert.Null(provider.GetKeyedService<GrpcChannel>(
|
||||
$"{SecretsGrpcHubClientExtensions.ChannelServiceKey}:fallback:0"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Registration_with_fallbacks_resolves_the_failover_reader_over_one_channel_per_endpoint()
|
||||
{
|
||||
ServiceCollection services = new();
|
||||
services.AddLogging();
|
||||
services.AddZbSecretsGrpcHubClient(HubConfig(new Dictionary<string, string?>
|
||||
{
|
||||
["Secrets:GrpcHub:FallbackEndpoints:0"] = "https://central-b:5001",
|
||||
["Secrets:GrpcHub:FallbackEndpoints:1"] = "https://central-c:5001",
|
||||
}));
|
||||
|
||||
using ServiceProvider provider = services.BuildServiceProvider();
|
||||
|
||||
Assert.IsType<FailoverSecretsHubReader>(provider.GetRequiredService<ISecretsHubReader>());
|
||||
|
||||
GrpcChannel primary = provider.GetRequiredKeyedService<GrpcChannel>(
|
||||
SecretsGrpcHubClientExtensions.ChannelServiceKey);
|
||||
GrpcChannel fallback0 = provider.GetRequiredKeyedService<GrpcChannel>(
|
||||
$"{SecretsGrpcHubClientExtensions.ChannelServiceKey}:fallback:0");
|
||||
GrpcChannel fallback1 = provider.GetRequiredKeyedService<GrpcChannel>(
|
||||
$"{SecretsGrpcHubClientExtensions.ChannelServiceKey}:fallback:1");
|
||||
|
||||
// Each channel dials its own endpoint — three distinct singletons, container-owned.
|
||||
Assert.Equal("central-a:5001", primary.Target);
|
||||
Assert.Equal("central-b:5001", fallback0.Target);
|
||||
Assert.Equal("central-c:5001", fallback1.Target);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Registration_refuses_a_malformed_fallback_endpoint_at_startup()
|
||||
{
|
||||
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() =>
|
||||
new ServiceCollection().AddZbSecretsGrpcHubClient(HubConfig(new Dictionary<string, string?>
|
||||
{
|
||||
["Secrets:GrpcHub:FallbackEndpoints:0"] = "not a uri",
|
||||
})));
|
||||
|
||||
Assert.Contains("Secrets:GrpcHub:FallbackEndpoints", ex.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
// A minimal valid hub-client configuration, extendable per test.
|
||||
private static IConfiguration HubConfig(Dictionary<string, string?>? extra = null)
|
||||
{
|
||||
var values = new Dictionary<string, string?>
|
||||
{
|
||||
["Secrets:GrpcHub:Endpoint"] = "https://central-a:5001",
|
||||
["Secrets:GrpcHub:BearerToken"] = "token",
|
||||
};
|
||||
|
||||
foreach ((string key, string? value) in extra ?? [])
|
||||
{
|
||||
values[key] = value;
|
||||
}
|
||||
|
||||
return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
|
||||
}
|
||||
|
||||
// Runs the hosted service through one startup sweep and shuts it down, asserting that nothing
|
||||
// escaped ExecuteAsync. The long sync interval means exactly one sweep happens.
|
||||
private async Task RunOneStartupSweepAsync(Func<bool> sweepObserved)
|
||||
|
||||
+60
@@ -75,6 +75,8 @@ public sealed class SecretsGrpcHubClientOptionsTests
|
||||
o => o.BearerToken = string.Empty,
|
||||
o => o.SyncInterval = TimeSpan.Zero,
|
||||
o => o.CallDeadline = TimeSpan.FromSeconds(-1),
|
||||
o => o.FallbackEndpoints.Add("not a uri"),
|
||||
o => o.FallbackEndpoints.Add("https://central:5001"),
|
||||
];
|
||||
|
||||
[Theory]
|
||||
@@ -116,6 +118,64 @@ public sealed class SecretsGrpcHubClientOptionsTests
|
||||
Assert.Contains("Secrets:GrpcHub:CallDeadline", ex.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Zero_fallback_endpoints_is_valid_and_the_default()
|
||||
{
|
||||
SecretsGrpcHubClientOptions options = Valid();
|
||||
|
||||
Assert.Empty(options.FallbackEndpoints);
|
||||
options.Validate();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_well_formed_fallback_endpoint_validates()
|
||||
{
|
||||
SecretsGrpcHubClientOptions options = Valid();
|
||||
options.FallbackEndpoints.Add("https://central-b:5001");
|
||||
options.FallbackEndpoints.Add("http://central-c:5001");
|
||||
|
||||
options.Validate();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData(" ")]
|
||||
[InlineData("central-b:5001")]
|
||||
[InlineData("/relative/path")]
|
||||
[InlineData("ftp://central-b:5001")]
|
||||
public void A_fallback_that_is_not_an_absolute_http_uri_names_its_key(string fallback)
|
||||
{
|
||||
SecretsGrpcHubClientOptions options = Valid();
|
||||
options.FallbackEndpoints.Add(fallback);
|
||||
|
||||
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
|
||||
Assert.Contains("Secrets:GrpcHub:FallbackEndpoints", ex.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("https://central:5001")]
|
||||
[InlineData("HTTPS://CENTRAL:5001")]
|
||||
public void A_fallback_duplicating_the_primary_endpoint_names_its_key(string fallback)
|
||||
{
|
||||
// A duplicated endpoint silently halves the failover the operator thinks they configured.
|
||||
SecretsGrpcHubClientOptions options = Valid();
|
||||
options.FallbackEndpoints.Add(fallback);
|
||||
|
||||
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
|
||||
Assert.Contains("Secrets:GrpcHub:FallbackEndpoints", ex.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_fallback_duplicating_another_fallback_names_its_key()
|
||||
{
|
||||
SecretsGrpcHubClientOptions options = Valid();
|
||||
options.FallbackEndpoints.Add("https://central-b:5001");
|
||||
options.FallbackEndpoints.Add("HTTPS://central-B:5001");
|
||||
|
||||
InvalidOperationException ex = Assert.Throws<InvalidOperationException>(() => options.Validate());
|
||||
Assert.Contains("Secrets:GrpcHub:FallbackEndpoints", ex.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Messages_name_the_section_the_caller_actually_bound()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user