dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(controlplane): a PSK secret deleted from the store revokes the key; bound refresh concurrency

Browse Source 
RefreshAsync now tells "the store answered: no such secret" (revocation - drop the refreshed key AND the
lazily cached one) apart from "the read failed" (keep the previous key). Store reads fan out through
Parallel.ForEachAsync capped at MaxConcurrentRefreshReads = 8 instead of one read per cluster at once.

Refs #733
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 13:49:56 -04:00
parent 2adeacc655
commit 896aaa2b02
2 changed files with 136 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs
+65 -25
View File
@@ -110,7 +110,8 @@ public interface IMeshPskProvider
/// <summary>
/// Re-reads the keys for <paramref name="clusterIds"/> from the store OFF the request path and
/// swaps them into the cache atomically (R03:3.4). Until the refresh completes, lookups keep
/// serving the previous key; a cluster whose read fails keeps its previous key too. Replaces
/// serving the previous key; a cluster whose read FAILS keeps its previous key too, while a cluster whose
/// secret the store reports ABSENT is revoked (no key from then on). Replaces
/// "invalidate on every refresh", which made the first inbound ack per cluster per
/// <c>ContactRefreshSeconds</c> block a gRPC thread on a synchronous, 3 s-bounded store read.
/// </summary>
@@ -331,45 +332,76 @@ public sealed class MeshPskProvider : IMeshPskProvider
_refreshed = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
}
/// <summary>
/// Ceiling on concurrent store reads during one <see cref="RefreshAsync"/>. A fleet of N clusters used to
/// fan out N simultaneous reads every contact refresh; the store (and anything behind it) sees at most
/// this many at once now.
/// </summary>
public const int MaxConcurrentRefreshReads = 8;
/// <inheritdoc />
public async Task RefreshAsync(IReadOnlyCollection<string> clusterIds, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(clusterIds);
var previous = _refreshed;
var next = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
var ids = clusterIds.Where(id => !string.IsNullOrWhiteSpace(id)).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
var results = new ConcurrentDictionary<string, StoreRead>(StringComparer.OrdinalIgnoreCase);
// Resolve every cluster concurrently — each read bounded like the synchronous path — and carry a
// failed cluster's PREVIOUS refreshed key forward: a transient store fault must not turn into a
// window in which that cluster has no key at all.
var reads = ids.Select(async id =>
// Bounded fan-out; each read bounded like the synchronous path.
await Parallel.ForEachAsync(
ids,
new ParallelOptions { MaxDegreeOfParallelism = MaxConcurrentRefreshReads, CancellationToken = ct },
async (id, token) => results[id] = await ResolveFromStoreAsync(id, token).ConfigureAwait(false))
.ConfigureAwait(false);
// Three outcomes per cluster (fix round 1 / I8):
// - Found → the new key;
// - NotFound → the store answered that the secret does not exist: a REVOCATION — the cluster gets no
// key (and its lazily cached one is evicted below), so a deleted secret stops working
// within one refresh instead of living on for the life of the process;
// - Failed → the store could not be read (threw / timed out / unnameable): keep the PREVIOUS key —
// a transient fault must not open a window in which the cluster has no key at all.
var next = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
var answered = new List<string>();
foreach (var id in ids)
{
var value = await ResolveFromStoreAsync(id, ct).ConfigureAwait(false);
return (Id: id, Value: value);
}).ToList();
foreach (var (id, value) in await Task.WhenAll(reads).ConfigureAwait(false))
{
if (!string.IsNullOrWhiteSpace(value)) next[id] = value;
else if (previous.TryGetValue(id, out var kept)) next[id] = kept;
var read = results[id];
switch (read.Outcome)
{
case StoreReadOutcome.Found:
next[id] = read.Value!;
answered.Add(id);
break;
case StoreReadOutcome.NotFound:
answered.Add(id);
break;
default:
if (previous.TryGetValue(id, out var kept)) next[id] = kept;
break;
}
}
// The swap. Then drop the lazy entries for the clusters now covered, so a key rotated between
// refreshes is served from the new snapshot rather than from a stale lazy resolution; entries
// for clusters NOT covered (their read failed and they had no previous value) are left alone.
// The swap. Then drop the lazy entries for every cluster the store ANSWERED for — found (serve the new
// snapshot, not a stale lazy resolution) or not found (revoked: a lazily cached key must die too).
// Entries for clusters whose read failed are left alone.
_refreshed = next;
foreach (var id in next.Keys) _cache.TryRemove(id, out _);
foreach (var id in answered) _cache.TryRemove(id, out _);
}
private enum StoreReadOutcome { Found, NotFound, Failed }
private readonly record struct StoreRead(StoreReadOutcome Outcome, string? Value);
/// <summary>
/// The asynchronous twin of <see cref="ResolveFromStore"/>: same name derivation, same bound, same
/// "anything but a usable key is null" contract — awaited instead of blocked on, because this one
/// runs on a refresh rather than inside a gRPC interceptor.
/// The asynchronous twin of <see cref="ResolveFromStore"/>: same name derivation, same bound — awaited
/// instead of blocked on, because this one runs on a refresh rather than inside a gRPC interceptor —
/// but it keeps "the secret does not exist" (a blank answer) apart from "the store could not be read".
/// </summary>
/// <param name="clusterId">The application <c>Cluster</c> id.</param>
/// <param name="ct">The caller's token, linked under the store-read bound.</param>
/// <returns>The stored key, or <c>null</c>.</returns>
private async Task<string?> ResolveFromStoreAsync(string clusterId, CancellationToken ct)
/// <returns>The outcome and, when found, the key.</returns>
private async Task<StoreRead> ResolveFromStoreAsync(string clusterId, CancellationToken ct)
{
SecretName name;
try
@@ -384,14 +416,22 @@ public sealed class MeshPskProvider : IMeshPskProvider
+ "refreshed from the store. Supply it through MeshTransport:ClusterPsks instead, or "
+ "rename the cluster to [a-z0-9._/-].",
clusterId);
return null;
return new StoreRead(StoreReadOutcome.Failed, null);
}
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(StoreReadTimeout);
try
{
return await _secretResolver.GetAsync(name, cts.Token).ConfigureAwait(false);
var value = await _secretResolver.GetAsync(name, cts.Token).ConfigureAwait(false);
if (!string.IsNullOrWhiteSpace(value)) return new StoreRead(StoreReadOutcome.Found, value);
_logger.LogWarning(
"Mesh PSK secret {SecretName} for cluster {ClusterId} is absent (or blank) in the store; the key is "
+ "REVOKED — commands to that cluster and acks from it are refused until it is restored.",
name.Value,
clusterId);
return new StoreRead(StoreReadOutcome.NotFound, null);
}
catch (Exception ex)
{
@@ -401,7 +441,7 @@ public sealed class MeshPskProvider : IMeshPskProvider
+ "previously resolved key (if any) stays in service until the next refresh.",
clusterId,
name.Value);
return null;
return new StoreRead(StoreReadOutcome.Failed, null);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs
+71
View File
@@ -78,6 +78,77 @@ public sealed class MeshPskProviderRefreshTests
psk.ShouldBe("k-1");
}
/// <summary>
/// Fix round 1 / I8 — "the store says there is no such secret" is a REVOCATION, not a read failure:
/// the previous key must stop being served (from the refreshed snapshot AND the lazy cache). A thrown
/// read keeps the previous key (the test above); a definitive not-found drops it.
/// </summary>
[Fact]
public async Task A_secret_deleted_from_the_store_revokes_the_key_on_the_next_refresh()
{
var resolver = new GatedResolver { Value = "k-1" };
var provider = Provider(resolver);
await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None);
provider.Matches("SITE-A", "k-1").ShouldBeTrue("positive control: the key was in service");
resolver.Value = null; // the secret was deleted
await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None);
provider.TryGetPsk("SITE-A", out _).ShouldBeFalse();
provider.Matches("SITE-A", "k-1").ShouldBeFalse();
}
/// <summary>The same revocation reaches a key that was first served from the LAZY cache, not a refresh.</summary>
[Fact]
public async Task A_deleted_secret_also_evicts_a_lazily_cached_key()
{
var resolver = new GatedResolver { Value = "k-lazy" };
var provider = Provider(resolver);
provider.TryGetPsk("SITE-A", out var lazy).ShouldBeTrue();
lazy.ShouldBe("k-lazy");
resolver.Value = null;
await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None);
provider.TryGetPsk("SITE-A", out _).ShouldBeFalse();
}
/// <summary>Fix round 1 minor — the refresh fans out to the store with BOUNDED concurrency, not one read per cluster at once.</summary>
[Fact]
public async Task Refresh_store_reads_are_concurrency_bounded()
{
var resolver = new ConcurrencyTrackingResolver();
var provider = Provider(resolver);
var clusters = Enumerable.Range(0, 40).Select(i => $"SITE-{i}").ToList();
await provider.RefreshAsync(clusters, CancellationToken.None);
resolver.Calls.ShouldBe(40);
resolver.MaxConcurrent.ShouldBeLessThanOrEqualTo(MeshPskProvider.MaxConcurrentRefreshReads);
resolver.MaxConcurrent.ShouldBeGreaterThan(1, "still concurrent, just bounded");
foreach (var c in clusters) provider.TryGetPsk(c, out _).ShouldBeTrue();
}
private sealed class ConcurrencyTrackingResolver : ISecretResolver
{
private int _inFlight;
private int _max;
private int _calls;
public int MaxConcurrent => Volatile.Read(ref _max);
public int Calls => Volatile.Read(ref _calls);
public async Task<string?> GetAsync(SecretName name, CancellationToken ct)
{
Interlocked.Increment(ref _calls);
var now = Interlocked.Increment(ref _inFlight);
int seen;
while (now > (seen = Volatile.Read(ref _max)) && Interlocked.CompareExchange(ref _max, now, seen) != seen) { }
await Task.Delay(20, ct);
Interlocked.Decrement(ref _inFlight);
return "k-" + name.Value;
}
}
/// <summary>A cluster that was never refreshed still resolves lazily on the request path (unchanged behaviour).</summary>
[Fact]
public async Task A_cluster_outside_the_refresh_set_still_resolves_lazily()