fix(controlplane): evict retired mesh channels and refresh PSKs off the request path

- MeshGrpcCommandDialer.RetainOnly (IMeshCommandDialer, default no-op): CentralCommunicationActor
  hands the dialer the current endpoint set on every contact refresh, so a renamed host or moved
  port releases its cached GrpcChannel and idle HTTP/2 connection instead of living for the
  process (R03:1.3).
- IMeshPskProvider.RefreshAsync (default = the old InvalidateAll): MeshPskProvider pre-resolves
  every known cluster's key concurrently, bounded like the sync path, and swaps an immutable
  snapshot in atomically; a failed read keeps that cluster's previous key. The actor calls it from
  ContactsLoaded (after SetKnownClusters, off-actor via PipeTo) instead of InvalidateAll, so the
  first ack per cluster per ContactRefreshSeconds no longer blocks a gRPC thread on a 3 s-bounded
  store read or gets denied past the bound (R03:3.4).

Refs #733

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs

@@ -201,6 +201,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
             LoadContactsFromDb();
         });
         Receive<ContactsLoaded>(HandleContactsLoaded);
+        Receive<PskRefreshed>(msg => _log.Debug("Mesh PSK refresh completed for {Count} cluster(s)", msg.ClusterCount));
+        Receive<PskRefreshFailed>(msg => _log.Warning(msg.Error,
+            "Mesh PSK refresh failed; previously resolved keys stay in service until the next refresh"));
         Receive<CommandDispatched>(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
     /// <summary>Internal tick asking for a contact-point refresh.</summary>
     public sealed record RefreshContacts;
 
+    /// <summary>The asynchronous PSK refresh (R03:3.4) completed.</summary>
+    /// <param name="ClusterCount">How many clusters were refreshed.</param>
+    public sealed record PskRefreshed(int ClusterCount);
+
+    /// <summary>The asynchronous PSK refresh (R03:3.4) threw; the previous keys stay in service.</summary>
+    /// <param name="Error">The failure.</param>
+    public sealed record PskRefreshFailed(Exception Error);
+
     /// <summary>Result of a contact-point load, piped back from the DB read.</summary>
     /// <param name="Targets">
     ///     The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a

src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs

@@ -106,6 +106,26 @@ public interface IMeshPskProvider
 
     /// <summary>Drops every cached key.</summary>
     void InvalidateAll();
+
+    /// <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
+    ///     "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>
+    /// <param name="clusterIds">The clusters central currently knows (the same set given to <see cref="SetKnownClusters"/>).</param>
+    /// <param name="ct">Cancels the refresh.</param>
+    /// <returns>Completes when the swap has happened.</returns>
+    /// <remarks>
+    ///     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.
+    /// </remarks>
+    Task RefreshAsync(IReadOnlyCollection<string> clusterIds, CancellationToken ct)
+    {
+        InvalidateAll();
+        return Task.CompletedTask;
+    }
 }
 
 /// <inheritdoc />
@@ -144,6 +164,16 @@ public sealed class MeshPskProvider : IMeshPskProvider
     private readonly ConcurrentDictionary<string, Lazy<string?>> _cache =
         new(StringComparer.OrdinalIgnoreCase);
 
+    /// <summary>
+    ///     Keys pre-resolved by <see cref="RefreshAsync"/>, consulted after the config map and before the
+    ///     lazy <see cref="_cache"/>. 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".
+    /// </summary>
+    private volatile IReadOnlyDictionary<string, string> _refreshed =
+        new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
+
     /// <summary>
     ///     The known-cluster gate's membership set, or <see langword="null"/> when the gate is
     ///     inactive (never published, or published empty — see <see cref="SetKnownClusters"/>).
@@ -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<string?>(
@@ -288,7 +325,85 @@ public sealed class MeshPskProvider : IMeshPskProvider
     }
 
     /// <inheritdoc />
-    public void InvalidateAll() => _cache.Clear();
+    public void InvalidateAll()
+    {
+        _cache.Clear();
+        _refreshed = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
+    }
+
+    /// <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();
+
+        // 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 _);
+    }
+
+    /// <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.
+    /// </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)
+    {
+        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;
+        }
+    }
 
     /// <summary>
     ///     Reads <c>OT-MESH-PSK-{clusterId}</c> from the Secrets store, converting every failure mode

src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs

@@ -34,6 +34,15 @@ public interface IMeshCommandDialer
     ///     <see cref="StatusCode.Unauthenticated"/> rather than dialled unauthenticated.
     /// </exception>
     Task DispatchAsync(string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct);
+
+    /// <summary>
+    ///     Releases any per-endpoint resources for endpoints NOT in <paramref name="nodeEndpoints"/>
+    ///     (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.
+    /// </summary>
+    /// <param name="nodeEndpoints">The endpoints still in the dispatch set.</param>
+    void RetainOnly(IReadOnlyCollection<string> nodeEndpoints) { }
 }
 
 /// <inheritdoc />
@@ -181,6 +190,30 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable
         await call.ResponseAsync.ConfigureAwait(false);
     }
 
+    /// <inheritdoc />
+    public void RetainOnly(IReadOnlyCollection<string> nodeEndpoints)
+    {
+        ArgumentNullException.ThrowIfNull(nodeEndpoints);
+        if (_disposed) return;
+
+        var keep = nodeEndpoints as IReadOnlySet<string> ?? 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);
+            }
+        }
+    }
+
     /// <summary>Marks the dialer disposed and disposes every cached gRPC channel.</summary>
     public void Dispose()
     {

tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs

@@ -930,6 +930,34 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
     /// <summary>A spawned gRPC-mode actor plus the two seams the tests observe it through.</summary>
     private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk);
 
+    /// <summary>
+    ///     R03:1.3 — every contact refresh hands the dialer the CURRENT endpoint set so it can release the
+    ///     channels of endpoints that left it.
+    /// </summary>
+    [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<OtOpcUaConfigDbContext> factory,
         RecordingDialer? dialer = null,
@@ -1014,6 +1042,19 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
                 : Task.CompletedTask;
         }
 
+        /// <summary>Every endpoint set handed to <see cref="RetainOnly"/>, in order.</summary>
+        public IReadOnlyList<IReadOnlyCollection<string>> RetainCalls
+        {
+            get { lock (_gate) return _retainCalls.ToList(); }
+        }
+
+        private readonly List<IReadOnlyCollection<string>> _retainCalls = [];
+
+        public void RetainOnly(IReadOnlyCollection<string> nodeEndpoints)
+        {
+            lock (_gate) _retainCalls.Add(nodeEndpoints.ToList());
+        }
+
         /// <summary>One recorded dial.</summary>
         public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId);
     }

tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs

@@ -224,6 +224,35 @@ public sealed class MeshGrpcCommandDialerTests
         Should.NotThrow(dialer.Dispose);
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<MeshGrpcCommandDialer>.Instance);
+        foreach (var endpoint in new[] { "http://127.0.0.1:1", "http://127.0.0.1:2" })
+        {
+            await Should.ThrowAsync<RpcException>(
+                () => 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<RpcException>(
+            () => 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()
     {

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;
+
+/// <summary>
+///     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. <see cref="MeshPskProvider.RefreshAsync"/> pre-resolves the
+///     keys off the request path and swaps them in atomically.
+/// </summary>
+public sealed class MeshPskProviderRefreshTests
+{
+    private static MeshPskProvider Provider(ISecretResolver resolver) =>
+        new(Options.Create(new MeshTransportOptions()), resolver, NullLogger<MeshPskProvider>.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();
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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");
+    }
+
+    /// <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()
+    {
+        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<string?> 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;
+        }
+    }
+}