diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs index bdab7ba7..bad1956c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs @@ -110,7 +110,8 @@ public interface IMeshPskProvider /// /// Re-reads the keys for 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 /// ContactRefreshSeconds block a gRPC thread on a synchronous, 3 s-bounded store read. /// @@ -331,45 +332,76 @@ public sealed class MeshPskProvider : IMeshPskProvider _refreshed = new Dictionary(StringComparer.OrdinalIgnoreCase); } + /// + /// Ceiling on concurrent store reads during one . 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. + /// + public const int MaxConcurrentRefreshReads = 8; + /// public async Task RefreshAsync(IReadOnlyCollection clusterIds, CancellationToken ct) { ArgumentNullException.ThrowIfNull(clusterIds); var previous = _refreshed; - var next = new Dictionary(StringComparer.OrdinalIgnoreCase); var ids = clusterIds.Where(id => !string.IsNullOrWhiteSpace(id)).Distinct(StringComparer.OrdinalIgnoreCase).ToList(); + var results = new ConcurrentDictionary(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(StringComparer.OrdinalIgnoreCase); + var answered = new List(); + 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); + /// - /// The asynchronous twin of : 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 : 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". /// /// The application Cluster id. /// The caller's token, linked under the store-read bound. - /// The stored key, or null. - private async Task ResolveFromStoreAsync(string clusterId, CancellationToken ct) + /// The outcome and, when found, the key. + private async Task 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); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs index e727c153..34129235 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs @@ -78,6 +78,77 @@ public sealed class MeshPskProviderRefreshTests psk.ShouldBe("k-1"); } + /// + /// 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. + /// + [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(); + } + + /// The same revocation reaches a key that was first served from the LAZY cache, not a refresh. + [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(); + } + + /// Fix round 1 minor — the refresh fans out to the store with BOUNDED concurrency, not one read per cluster at once. + [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 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; + } + } + /// A cluster that was never refreshed still resolves lazily on the request path (unchanged behaviour). [Fact] public async Task A_cluster_outside_the_refresh_set_still_resolves_lazily()