diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs index cb1ea253..16f7b5d4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs @@ -201,6 +201,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers LoadContactsFromDb(); }); Receive(HandleContactsLoaded); + Receive(msg => _log.Debug("Mesh PSK refresh completed for {Count} cluster(s)", msg.ClusterCount)); + Receive(msg => _log.Warning(msg.Error, + "Mesh PSK refresh failed; previously resolved keys stay in service until the next refresh")); Receive(HandleCommandDispatched); // A faulted LoadContactsFromDb task is piped here as a Status.Failure. Without this handler @@ -710,6 +713,20 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers _targets = msg.Targets; + // R03:1.3 — release the channels of endpoints that left the dispatch set (host renamed, port + // moved). Only the resolved dialer is asked; nothing is resolved just to prune it. + if (_dialer is { } dialer) + { + try + { + dialer.RetainOnly(_targets.Select(t => t.Endpoint).ToHashSet(StringComparer.Ordinal)); + } + catch (Exception ex) + { + _log.Warning(ex, "Mesh dialer failed to release channels for retired endpoints; they stay cached until the next refresh"); + } + } + if (_targets.Count == 0) { _log.Warning( @@ -780,7 +797,37 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers } pskProvider.SetKnownClusters(msg.ClusterIds); - pskProvider.InvalidateAll(); + + // R03:3.4 — refresh the per-cluster keys OFF the request path and swap them in atomically, + // instead of dropping the cache and making the first ack per cluster re-read the store on a + // gRPC thread. A provider that does not pre-resolve (the interface default, and every fake) + // still drops its cache here, synchronously, exactly as before. + CancellationToken ct; + try + { + ct = _lifecycle.Token; + } + catch (ObjectDisposedException) + { + return; // Stopping. + } + var self = Self; + Task refresh; + try + { + refresh = pskProvider.RefreshAsync(msg.ClusterIds, ct); + } + catch (Exception ex) + { + _log.Warning(ex, "Mesh PSK refresh threw synchronously; cached keys are unchanged until the next refresh"); + refresh = Task.CompletedTask; + } + if (!refresh.IsCompletedSuccessfully) + { + refresh.PipeTo(self, + success: () => new PskRefreshed(msg.ClusterIds.Count), + failure: ex => new PskRefreshFailed(ex)); + } if (!_gatePublished) { @@ -795,6 +842,14 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// Internal tick asking for a contact-point refresh. public sealed record RefreshContacts; + /// The asynchronous PSK refresh (R03:3.4) completed. + /// How many clusters were refreshed. + public sealed record PskRefreshed(int ClusterCount); + + /// The asynchronous PSK refresh (R03:3.4) threw; the previous keys stay in service. + /// The failure. + public sealed record PskRefreshFailed(Exception Error); + /// Result of a contact-point load, piped back from the DB read. /// /// The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a 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 a7282694..bdab7ba7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs @@ -106,6 +106,26 @@ public interface IMeshPskProvider /// Drops every cached key. void InvalidateAll(); + + /// + /// 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 + /// "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. + /// + /// The clusters central currently knows (the same set given to ). + /// Cancels the refresh. + /// Completes when the swap has happened. + /// + /// The default implementation is the pre-R03:3.4 behaviour — drop the cache and let the next lookup + /// re-read — so a fake or a non-caching provider needs nothing more. + /// + Task RefreshAsync(IReadOnlyCollection clusterIds, CancellationToken ct) + { + InvalidateAll(); + return Task.CompletedTask; + } } /// @@ -144,6 +164,16 @@ public sealed class MeshPskProvider : IMeshPskProvider private readonly ConcurrentDictionary> _cache = new(StringComparer.OrdinalIgnoreCase); + /// + /// Keys pre-resolved by , consulted after the config map and before the + /// lazy . Swapped wholesale (volatile reference to an immutable dictionary), + /// never mutated, so a lookup racing a refresh sees the whole previous snapshot or the whole next + /// one. A cluster missing from it — never refreshed, or whose read failed — falls through to the + /// lazy path exactly as before, which is what keeps "refresh failed" from meaning "no key". + /// + private volatile IReadOnlyDictionary _refreshed = + new Dictionary(StringComparer.OrdinalIgnoreCase); + /// /// The known-cluster gate's membership set, or when the gate is /// inactive (never published, or published empty — see ). @@ -210,6 +240,13 @@ public sealed class MeshPskProvider : IMeshPskProvider return true; } + // Pre-resolved by the asynchronous refresh: no store read on this (gRPC request) thread at all. + if (_refreshed.TryGetValue(clusterId, out var refreshed)) + { + psk = refreshed; + return true; + } + var entry = _cache.GetOrAdd( clusterId, static (id, self) => new Lazy( @@ -288,7 +325,85 @@ public sealed class MeshPskProvider : IMeshPskProvider } /// - public void InvalidateAll() => _cache.Clear(); + public void InvalidateAll() + { + _cache.Clear(); + _refreshed = new Dictionary(StringComparer.OrdinalIgnoreCase); + } + + /// + 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(); + + // 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 => + { + 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; + } + + // 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. + _refreshed = next; + foreach (var id in next.Keys) _cache.TryRemove(id, out _); + } + + /// + /// 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 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) + { + SecretName name; + try + { + name = new SecretName(SecretNamePrefix + clusterId); + } + catch (ArgumentException ex) + { + _logger.LogWarning( + ex, + "Cluster {ClusterId} cannot be expressed as a secret name, so its mesh PSK cannot be " + + "refreshed from the store. Supply it through MeshTransport:ClusterPsks instead, or " + + "rename the cluster to [a-z0-9._/-].", + clusterId); + return null; + } + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + cts.CancelAfter(StoreReadTimeout); + try + { + return await _secretResolver.GetAsync(name, cts.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "Failed to refresh the mesh PSK for cluster {ClusterId} from secret {SecretName}; the " + + "previously resolved key (if any) stays in service until the next refresh.", + clusterId, + name.Value); + return null; + } + } /// /// Reads OT-MESH-PSK-{clusterId} from the Secrets store, converting every failure mode diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs index a10d228a..31a19b13 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs @@ -34,6 +34,15 @@ public interface IMeshCommandDialer /// rather than dialled unauthenticated. /// Task DispatchAsync(string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct); + + /// + /// Releases any per-endpoint resources for endpoints NOT in + /// (R03:1.3). Central calls it on every contact refresh with the current dispatch set, so a + /// renamed host or moved port does not leave the old channel and its idle HTTP/2 connection + /// alive for the life of the process. Default no-op for dialers that cache nothing. + /// + /// The endpoints still in the dispatch set. + void RetainOnly(IReadOnlyCollection nodeEndpoints) { } } /// @@ -181,6 +190,30 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable await call.ResponseAsync.ConfigureAwait(false); } + /// + public void RetainOnly(IReadOnlyCollection nodeEndpoints) + { + ArgumentNullException.ThrowIfNull(nodeEndpoints); + if (_disposed) return; + + var keep = nodeEndpoints as IReadOnlySet ?? nodeEndpoints.ToHashSet(StringComparer.Ordinal); + foreach (var endpoint in _channels.Keys) + { + if (keep.Contains(endpoint)) continue; + // A dial that already took this channel out of the cache before the removal fails once, as + // a named per-node RpcException the actor logs; the next dial to that endpoint (if it ever + // returns to the set) builds a fresh channel. Bounded to the dials in flight at the instant + // of a contact refresh, the same accepted race Dispose documents. + if (_channels.TryRemove(endpoint, out var channel)) + { + channel.Dispose(); + _log.LogInformation( + "Mesh dialer released the cached channel for {Endpoint}: it is no longer in the dispatch set", + endpoint); + } + } + } + /// Marks the dialer disposed and disposes every cached gRPC channel. public void Dispose() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs index 9843c8fe..51dcd046 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs @@ -930,6 +930,34 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase /// A spawned gRPC-mode actor plus the two seams the tests observe it through. private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk); + /// + /// R03:1.3 — every contact refresh hands the dialer the CURRENT endpoint set so it can release the + /// channels of endpoints that left it. + /// + [Fact] + public void Each_contact_refresh_tells_the_dialer_which_endpoints_remain() + { + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false), ("host-b:4053", "host-b", 4056, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + // The dialer is resolved lazily on first use; a dispatch forces that, and the NEXT refresh prunes. + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); + + using (var db = factory.CreateDbContext()) + { + db.ClusterNodes.Single(n => n.NodeId == "host-b:4053").MaintenanceMode = true; + db.SaveChanges(); + } + rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts()); + + AwaitCondition( + () => rig.Dialer.RetainCalls.Any(set => set.Count == 1 && set.Single().Contains("host-a")), + TimeSpan.FromSeconds(5)); + rig.Dialer.RetainCalls.Last().ShouldBe(new[] { "http://host-a:4056" }); + } + private GrpcRig SpawnGrpcRig( IDbContextFactory factory, RecordingDialer? dialer = null, @@ -1014,6 +1042,19 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase : Task.CompletedTask; } + /// Every endpoint set handed to , in order. + public IReadOnlyList> RetainCalls + { + get { lock (_gate) return _retainCalls.ToList(); } + } + + private readonly List> _retainCalls = []; + + public void RetainOnly(IReadOnlyCollection nodeEndpoints) + { + lock (_gate) _retainCalls.Add(nodeEndpoints.ToList()); + } + /// One recorded dial. public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs index 2d110eb6..fc177361 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs @@ -224,6 +224,35 @@ public sealed class MeshGrpcCommandDialerTests Should.NotThrow(dialer.Dispose); } + /// + /// R03:1.3 — a channel for an endpoint that left the dispatch set (host renamed, port moved) is + /// released on the next refresh instead of living, with its idle connection, for the life of central. + /// Positive control: a dial to that endpoint afterwards rebuilds a channel. + /// + [Fact] + public async Task RetainOnly_releases_channels_for_endpoints_that_left_the_dispatch_set() + { + var dialer = new MeshGrpcCommandDialer( + new FakePskProvider(("SITE-A", "psk-site-a")), + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 1 }), + NullLogger.Instance); + foreach (var endpoint in new[] { "http://127.0.0.1:1", "http://127.0.0.1:2" }) + { + await Should.ThrowAsync( + () => dialer.DispatchAsync(endpoint, Envelope(), "SITE-A", CancellationToken.None)); + } + dialer.CachedChannelCount.ShouldBe(2); + + dialer.RetainOnly(new[] { "http://127.0.0.1:1" }); + + dialer.CachedChannelCount.ShouldBe(1); + + await Should.ThrowAsync( + () => dialer.DispatchAsync("http://127.0.0.1:2", Envelope(), "SITE-A", CancellationToken.None)); + dialer.CachedChannelCount.ShouldBe(2, "a returning endpoint gets a fresh channel"); + Should.NotThrow(dialer.Dispose); + } + [Fact] public void It_activates_from_the_container_despite_the_optional_client_factory_parameter() { 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 new file mode 100644 index 00000000..e727c153 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs @@ -0,0 +1,129 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; +using ZB.MOM.WW.Secrets.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; + +/// +/// R03:3.4 — central invalidated every cached key on every contact refresh (default 60 s), so the +/// first inbound ack per cluster per minute blocked a gRPC thread on a synchronous, 3 s-bounded store +/// read and was denied past the bound. pre-resolves the +/// keys off the request path and swaps them in atomically. +/// +public sealed class MeshPskProviderRefreshTests +{ + private static MeshPskProvider Provider(ISecretResolver resolver) => + new(Options.Create(new MeshTransportOptions()), resolver, NullLogger.Instance); + + [Fact] + public async Task After_a_refresh_the_lookup_is_served_without_touching_the_store() + { + var resolver = new GatedResolver { Value = "k-1" }; + var provider = Provider(resolver); + + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + resolver.Calls.ShouldBe(1); + + provider.TryGetPsk("SITE-A", out var psk).ShouldBeTrue(); + psk.ShouldBe("k-1"); + resolver.Calls.ShouldBe(1, "the request path did not consult the store"); + provider.Matches("SITE-A", "k-1").ShouldBeTrue(); + } + + /// + /// While a refresh is in flight the PREVIOUS key is still served (no invalidate-then-reread window); + /// once it completes the rotated key is served. Positive control on both sides of the swap. + /// + [Fact] + public async Task A_rotated_key_is_swapped_in_atomically_with_the_old_key_served_until_then() + { + var resolver = new GatedResolver { Value = "old" }; + var provider = Provider(resolver); + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + provider.TryGetPsk("SITE-A", out var before).ShouldBeTrue(); + before.ShouldBe("old"); + + resolver.Value = "new"; + resolver.Hold(); + var refresh = provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + resolver.Entered.Wait(TimeSpan.FromSeconds(5)).ShouldBeTrue(); + + // Mid-refresh: the old key, from memory — the store is NOT re-read on this path. + var callsMidRefresh = resolver.Calls; + provider.TryGetPsk("SITE-A", out var during).ShouldBeTrue(); + during.ShouldBe("old"); + resolver.Calls.ShouldBe(callsMidRefresh); + + resolver.Release(); + await refresh; + provider.TryGetPsk("SITE-A", out var after).ShouldBeTrue(); + after.ShouldBe("new"); + } + + [Fact] + public async Task A_cluster_whose_refresh_fails_keeps_its_previous_key() + { + var resolver = new GatedResolver { Value = "k-1" }; + var provider = Provider(resolver); + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + + resolver.Throw = new InvalidOperationException("store offline"); + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + + provider.TryGetPsk("SITE-A", out var psk).ShouldBeTrue(); + psk.ShouldBe("k-1"); + } + + /// 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() + { + var resolver = new GatedResolver { Value = "k-b" }; + var provider = Provider(resolver); + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + + provider.TryGetPsk("SITE-B", out var psk).ShouldBeTrue(); + psk.ShouldBe("k-b"); + resolver.Calls.ShouldBe(2); + } + + [Fact] + public async Task InvalidateAll_drops_refreshed_keys_too() + { + var resolver = new GatedResolver { Value = "k-1" }; + var provider = Provider(resolver); + await provider.RefreshAsync(new[] { "SITE-A" }, CancellationToken.None); + + provider.InvalidateAll(); + resolver.Value = "k-2"; + provider.TryGetPsk("SITE-A", out var psk).ShouldBeTrue(); + psk.ShouldBe("k-2"); + } + + private sealed class GatedResolver : ISecretResolver + { + private int _calls; + private volatile TaskCompletionSource? _gate; + + public string? Value { get; set; } + public Exception? Throw { get; set; } + public int Calls => Volatile.Read(ref _calls); + public ManualResetEventSlim Entered { get; } = new(); + + public void Hold() => _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + public void Release() => _gate?.TrySetResult(); + + public async Task GetAsync(SecretName name, CancellationToken ct) + { + Interlocked.Increment(ref _calls); + Entered.Set(); + if (_gate is { } gate) await gate.Task.WaitAsync(ct); + if (Throw is not null) throw Throw; + return Value; + } + } +}