From 48dc2120c659867b5e49164ca1b94d7dc8e1ee4d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 06:44:20 -0400 Subject: [PATCH] fix(mesh): gate follows the endpoint not the mode + review fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The known-cluster gate published only under Mode=Grpc, but the interceptor it feeds is registered on every admin node and 1B maps CentralCommandService by LISTENER PORT — neither consults the mode. A dark-switch rollback to ClusterClient or Dps would therefore have left that endpoint serving with the gate permanently inactive, silently restoring both the timing oracle and the pre-auth secret-store/audit flood, as a side effect of a rollback nobody would connect to authentication. The gate is central's cluster inventory, not transport state, so it is now decoupled from the mode entirely: - SetKnownClusters + InvalidateAll run on EVERY successful contact load in EVERY mode, and the contact-refresh timer starts in all three. Under Dps this is a ConfigDb query every ContactRefreshSeconds that did not run before; central always has a ConfigDb, and one indexed read a minute is the cheaper half of the trade. - The GATE set is built from ALL ClusterNode rows. The eligibility filter moved out of the query's WHERE and into the loop, so the FAN set is byte-for-byte what it was (enabled && !maintenance) while the gate widens. This also dissolves the maintenance coupling flagged in 1A.2: a cluster whose nodes are all disabled or in maintenance can still ack, and refusing it would make maintenance mode an authentication switch. Central still refuses clusters it has never heard of. - HandleContactsLoaded keeps a documented order — targets adopted, gate published, cache dropped — now pinned by test. Guards: PreStart mirrors the null-dialer error for a null PSK provider (silent absence is the worst outcome there — nothing malfunctions, so nothing is investigated, while the gate never publishes and keys never rotate); a failed contact load now says out loud that the gate is stale and, before any load has succeeded, inactive; MeshGrpcCommandDialer clamps a non-positive DispatchTimeoutSeconds to 10 with a construction Warning, because the validator for that key is driver-role-scoped and a pure-admin central reads it unvalidated; the dialer's local PSK refusal was shortened so its remediation survives the actor's 200-char truncation of remote-authored status detail; the portless-node warn-once set is rebuilt each refresh, so a node that regains then loses a port warns again — the regression is the occurrence worth seeing. Docs/pins: dialer disposal comment no longer claims one-way; the two interceptor-test comments claiming Matches has a default implementation corrected (it deliberately has none); new pins for per-refresh invalidation, the SetKnownClusters-before-InvalidateAll order, and one channel per endpoint via a new CachedChannelCount. The_known_cluster_set_is_not_published_by_the_other_two_modes is replaced — its premise is obsolete — by three facts: every mode publishes, gate set is a strict superset of the fan set, and the handler ordering. ControlPlane.Tests 230 -> 236, Host.Tests 92, Cluster.Tests 184, 0 warnings. --- .../CentralCommunicationActor.cs | 162 +++++++++++++----- .../Communication/MeshGrpcCommandDialer.cs | 61 +++++-- .../CentralCommunicationActorTests.cs | 134 +++++++++++++-- .../MeshGrpcCommandDialerTests.cs | 85 ++++++++- .../CentralCommandAuthInterceptorTests.cs | 10 +- 5 files changed, 368 insertions(+), 84 deletions(-) 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 635194ff..60f3b7c2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs @@ -159,13 +159,23 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers // an operator cannot distinguish "no nodes configured" from "the database is down", and the // contact set silently freezes at whatever it last held. // - // Under Grpc that freeze extends to the known-cluster gate, deliberately: a failed load must - // neither widen the gate nor narrow it, and the previous set is the last thing central - // actually knew. - Receive(f => _log.Warning( - f.Cause, - "Failed to load ClusterNode contact points; the mesh contact set was NOT refreshed and " - + "may be stale or empty. Commands may be dropped until the next refresh")); + // The freeze extends to the known-cluster gate, deliberately: a failed load must neither + // widen the gate nor narrow it, and the previous set is the last thing central actually knew. + // The second line exists because that freeze is INVISIBLE otherwise, and it is fail-OPEN + // before the first successful load — a central that has never reached its ConfigDb refuses + // no cluster at all. + Receive(f => + { + _log.Warning( + f.Cause, + "Failed to load ClusterNode contact points; the mesh contact set was NOT refreshed " + + "and may be stale or empty. Commands may be dropped until the next refresh"); + _log.Warning( + "The known-cluster gate was NOT refreshed either: it still holds the set from the " + + "last successful load, and if none has succeeded yet (boot, or central SQL Server " + + "unreachable since start) it is INACTIVE — every cluster id reaches the secret " + + "store, which is the pre-authentication cost the gate exists to remove"); + }); Receive(_ => { /* DPS subscribe confirmation */ }); } @@ -177,8 +187,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers { _log.Info( "Mesh transport is {Mode}: commands are published on DistributedPubSub and no " - + "ClusterClient is created", _options.Mode); - return; + + "ClusterClient is created. The contact refresh still runs — it is what keeps the " + + "known-cluster gate current, and that gate belongs to the inbound ack endpoint, " + + "which is mapped by port and does not care about this mode", _options.Mode); } if (_useGrpc && _dialer is null) @@ -192,8 +203,25 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers + "IMeshCommandDialer"); } - // The refresh drives more than ClusterClient contacts under Grpc: it is also what publishes - // the dispatch target set and keeps the PSK provider's known-cluster gate current. + if (_pskProvider is null) + { + // Mode-independent, like the gate itself. A silent absence here is the worst of the + // three outcomes in this method: nothing malfunctions, so nothing is ever investigated, + // while the gate stays permanently inactive and a rotated key is never picked up. + _log.Error( + "No IMeshPskProvider was supplied to this actor; the known-cluster gate will NEVER " + + "be published and cached per-cluster keys will NEVER be dropped. Inbound mesh acks " + + "stay exposed to the pre-authentication secret-store cost the gate removes, and a " + + "rotated key needs a central restart. Check the ControlPlane registration"); + } + + // Started in EVERY mode. The refresh is not transport state: it is central's ClusterNode + // inventory, and the known-cluster gate it feeds protects an endpoint mapped by PORT, not by + // MeshTransport:Mode. Gating the timer on the mode would mean a dark-switch rollback to + // ClusterClient or Dps left that endpoint live with the gate permanently inactive — silently + // reopening both the timing oracle and the pre-auth store/audit flood. Under Dps this is a + // ConfigDb query every ContactRefreshSeconds that did not run before; central always has a + // ConfigDb, and one indexed read a minute is the cheaper half of the trade. Timers.StartPeriodicTimer( RefreshTimerKey, new RefreshContacts(), @@ -487,10 +515,18 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers Task.Run(async () => { await using var db = await dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false); + // UNFILTERED, with the eligibility flags projected and applied per row below. The two + // outputs of this read want different row sets: the contact/dispatch sets are the + // enabled, non-maintenance nodes central talks TO, while the known-cluster gate answers + // "is this a cluster central has ever heard of" and must include a cluster whose every + // node is disabled or in maintenance — such a cluster can still send an ack, and denying + // it would make maintenance mode an authentication switch. var rows = await db.ClusterNodes .AsNoTracking() - .Where(n => n.Enabled && !n.MaintenanceMode) - .Select(n => new { n.NodeId, n.ClusterId, n.Host, n.AkkaPort, n.GrpcPort }) + .Select(n => new + { + n.NodeId, n.ClusterId, n.Host, n.AkkaPort, n.GrpcPort, n.Enabled, n.MaintenanceMode, + }) .ToListAsync(ct) .ConfigureAwait(false); @@ -506,15 +542,21 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers var targets = new List(rows.Count); var portless = new List(); - // Every cluster with at least one enabled, non-maintenance row — the known-cluster gate's - // membership set. Deliberately independent of GrpcPort: the gate answers "is this a cluster - // central knows", and a cluster central cannot currently dial can still send it an ack. + // Every cluster with ANY ClusterNode row — the known-cluster gate's membership set, and + // deliberately a superset of the fan set. Independent of Enabled, MaintenanceMode and + // GrpcPort alike: none of those describe whether the cluster exists. var clusterIds = new HashSet(StringComparer.OrdinalIgnoreCase); foreach (var row in rows) { clusterIds.Add(row.ClusterId); + // The eligibility filter that used to be the query's WHERE clause. Everything below + // this line — receptionist contacts and gRPC targets — sees exactly the rows it saw + // before. + if (!row.Enabled || row.MaintenanceMode) + continue; + if (row.GrpcPort is { } grpcPort) targets.Add(new MeshNodeTarget( row.NodeId, row.ClusterId, $"http://{row.Host}:{grpcPort}")); @@ -548,11 +590,16 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers private void HandleContactsLoaded(ContactsLoaded msg) { + // Order is load-bearing and is pinned by test: the dispatch set is adopted BEFORE the gate is + // published, so anything that observes the gate can rely on the targets already being current. if (_useGrpc) - { - HandleGrpcContactsLoaded(msg); - return; - } + AdoptGrpcTargets(msg); + + // EVERY mode. See PreStart — the gate protects an endpoint mapped by port, not by mode. + PublishKnownClusters(msg); + + if (!_useClusterClient) + return; // Dps or Grpc: no ClusterClient to rebuild. foreach (var bad in msg.Malformed) { @@ -613,33 +660,28 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers } /// - /// Adopts a contact load under : the dispatch - /// target set, and the two things the PSK provider needs kept current. + /// Adopts a contact load's gRPC dispatch set. only + /// — the other two transports have no use for it. /// /// The load result. /// - /// - /// The malformed-receptionist-address warnings are deliberately NOT logged here: nothing - /// on this transport dials a receptionist, and reporting an unusable Akka address as a - /// command-plane problem would send an operator after the wrong column. A host that is - /// unusable as a gRPC endpoint surfaces instead as a named per-node dial failure on every - /// dispatch, which is a truer signal. - /// - /// - /// InvalidateAll, not per-cluster Invalidate. The requirement is that - /// a rotated key takes effect within one refresh, and only a whole-cache drop delivers - /// that: a key can rotate without the cluster's membership changing at all, so invalidating - /// only departed clusters would serve the pre-rotation key forever. The cost is one store - /// read per cluster per refresh — the cache holds a single entry per cluster — and it is - /// paid only by clusters that are actually dialled. - /// + /// The malformed-receptionist-address warnings are deliberately NOT logged here: nothing on + /// this transport dials a receptionist, and reporting an unusable Akka address as a + /// command-plane problem would send an operator after the wrong column. A host that is + /// unusable as a gRPC endpoint surfaces instead as a named per-node dial failure on every + /// dispatch, which is a truer signal. /// - private void HandleGrpcContactsLoaded(ContactsLoaded msg) + private void AdoptGrpcTargets(ContactsLoaded msg) { + // Rebuilt against this refresh's skip list, not merely added to. A node that gained a port + // and later lost it again must warn a second time — the operator's mental model is "the last + // refresh told me", not "you were told once, in a log that has since rotated away". + _warnedPortlessNodes.IntersectWith(msg.PortlessNodes); + foreach (var nodeId in msg.PortlessNodes) { - // Warn-once per node id: a persistently portless row is re-skipped every refresh, and the - // same repetition problem TelemetryNodeSource solved this way. + // Warn-once per node id per continuous skip run: a persistently portless row is + // re-skipped every refresh, and that is the repetition TelemetryNodeSource suppresses. if (_warnedPortlessNodes.Add(nodeId)) { _log.Warning( @@ -651,11 +693,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers _targets = msg.Targets; - // Published AFTER the target set is adopted, so anything observing the gate can rely on the - // dispatch set already being current. - _pskProvider?.SetKnownClusters(msg.ClusterIds); - _pskProvider?.InvalidateAll(); - if (_targets.Count == 0) { _log.Warning( @@ -665,6 +702,35 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers } } + /// + /// Hands the PSK provider this refresh's cluster inventory and drops its key cache. Runs in + /// every transport mode. + /// + /// The load result. + /// + /// + /// Not gated on , on purpose. The gate feeds + /// CentralCommandAuthInterceptor, which is registered on every admin node, and the + /// ack service it protects is mapped by listener port — neither consults the mode. Gating + /// this on Grpc would mean a dark-switch rollback to ClusterClient or + /// Dps left that endpoint serving with a permanently inactive gate: the timing + /// oracle and the pre-authentication store/audit flood would both come back, silently, + /// as a side effect of a rollback nobody would connect to authentication. + /// + /// + /// InvalidateAll, not per-cluster Invalidate. The requirement is that + /// a rotated key takes effect within one refresh, and only a whole-cache drop delivers + /// that: a key can rotate without the cluster's membership changing at all, so + /// invalidating only departed clusters would serve the pre-rotation key forever. The cost + /// is one store read per cluster per refresh — the cache holds a single entry per cluster. + /// + /// + private void PublishKnownClusters(ContactsLoaded msg) + { + _pskProvider?.SetKnownClusters(msg.ClusterIds); + _pskProvider?.InvalidateAll(); + } + /// Stops a cluster's existing client and creates a replacement for its new contact set. /// /// @@ -740,8 +806,10 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// the load task so the warn-once bookkeeping stays on the actor's thread. /// /// - /// Every application Cluster with at least one enabled, non-maintenance row — - /// the PSK provider's known-cluster gate membership. + /// Every application Cluster with any ClusterNode row — the PSK + /// provider's known-cluster gate membership, and deliberately a superset of + /// : a cluster whose nodes are all disabled or in maintenance can + /// still send an ack, and refusing it would make maintenance mode an authentication switch. /// public sealed record ContactsLoaded( IReadOnlyDictionary> ContactsByCluster, 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 46dcd830..10d9761f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs @@ -1,6 +1,7 @@ using System.Collections.Concurrent; using Grpc.Core; using Grpc.Net.Client; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; @@ -63,13 +64,21 @@ public interface IMeshCommandDialer /// caller's token so actor shutdown still cancels everything in flight. /// /// -/// Registered as a singleton and disposed by the container. Disposal is one-way: a dial -/// after it throws rather than silently repopulating the cache with channels nobody will -/// ever dispose. +/// Registered as a singleton and disposed by the container. A dial started after disposal +/// throws, so no new work repopulates the cache — but a dial already past that check can +/// still add a channel after the cache was cleared. That leak is bounded to the dials in +/// flight at shutdown, which is the only time disposal happens, so it is accepted rather +/// than serialized against. /// /// public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable { + /// + /// Substituted for a non-positive . + /// Deliberately the option's own default rather than a second value to reconcile. + /// + private const int FallbackDispatchTimeoutSeconds = 10; + /// /// Header naming the cluster the presented bearer belongs to. Central's own inbound gate /// reads the same header on the ack path; sending it outbound keeps one convention and tells @@ -78,6 +87,7 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable private const string ClusterHeader = "x-otopcua-cluster"; private readonly IMeshPskProvider _pskProvider; + private readonly ILogger _log; private readonly int _dispatchTimeoutSeconds; private readonly Func _clientFactory; private readonly ConcurrentDictionary _channels = new(StringComparer.Ordinal); @@ -87,23 +97,53 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable /// Initializes a new instance of the class. /// Supplies the bearer for the target node's cluster. /// Mesh transport options; supplies the per-call deadline. + /// Logger for the construction-time deadline check. /// /// Seam for tests: builds the generated client for an endpoint. Defaults to a real h2c /// , one cached per endpoint. /// + /// + /// A non-positive DispatchTimeoutSeconds is clamped, not honoured. Zero would mean an + /// unbounded call — the exact failure the deadline exists to prevent — and the validator that + /// rejects it is scoped to driver-role nodes, so the one deployment shape that reads this + /// value unvalidated is a pure-admin central. Clamping loudly is better than a fleet-wide fan + /// that can never time out. + /// public MeshGrpcCommandDialer( IMeshPskProvider pskProvider, IOptions options, + ILogger log, Func? clientFactory = null) { ArgumentNullException.ThrowIfNull(pskProvider); ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(log); _pskProvider = pskProvider; - _dispatchTimeoutSeconds = options.Value.DispatchTimeoutSeconds; + _log = log; + + var configured = options.Value.DispatchTimeoutSeconds; + if (configured <= 0) + { + _log.LogWarning( + "MeshTransport:DispatchTimeoutSeconds is {Configured}, which would leave every mesh " + + "command dial unbounded; using {Fallback}s instead. Set a positive value — the " + + "boot validator for this key covers driver-role nodes only, so a pure-admin central " + + "reaches this line rather than failing at startup.", + configured, FallbackDispatchTimeoutSeconds); + } + + _dispatchTimeoutSeconds = configured > 0 ? configured : FallbackDispatchTimeoutSeconds; _clientFactory = clientFactory ?? DefaultClientFactory; } + /// + /// Gets the number of endpoints holding a cached . Exposed so the + /// cache — the whole defence against a socket per call across a fleet-wide fan — is + /// assertable rather than merely intended. + /// + public int CachedChannelCount => _channels.Count; + /// public async Task DispatchAsync( string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct) @@ -117,12 +157,14 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable { // The cluster id is operator-authored configuration, not a secret, so naming it is what // makes this actionable. The key never appears here or anywhere else in this class. + // Kept short on purpose: the actor truncates a status detail at 200 characters because a + // REMOTE node authors it, and this locally-authored one has to survive that cap with its + // remediation intact. It measures ~160 characters with a six-character cluster id, so an + // id would have to exceed ~40 characters to lose the trailing rig hint. throw new RpcException(new Status( StatusCode.Unauthenticated, - $"No mesh pre-shared key could be established for cluster '{clusterId}'; the dial was " - + "refused locally rather than made unauthenticated. Seed the secret " - + "OT-MESH-PSK- on this node, or set MeshTransport:ClusterPsks: " - + "on a rig.")); + $"No mesh pre-shared key for cluster '{clusterId}'; dial refused locally. Seed secret " + + "OT-MESH-PSK-, or set MeshTransport:ClusterPsks: on a rig.")); } var headers = new Metadata @@ -132,8 +174,7 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable }; using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - if (_dispatchTimeoutSeconds > 0) - deadlineCts.CancelAfter(TimeSpan.FromSeconds(_dispatchTimeoutSeconds)); + deadlineCts.CancelAfter(TimeSpan.FromSeconds(_dispatchTimeoutSeconds)); var client = _clientFactory(nodeEndpoint); using var call = client.ExecuteAsync(envelope, headers, cancellationToken: deadlineCts.Token); 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 8563f781..a3df8f20 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 @@ -461,6 +461,28 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase rig.Dialer.Calls[0].Endpoint.ShouldBe("http://host-b:4056"); } + [Fact] + public void A_node_that_regains_then_loses_its_GrpcPort_warns_again() + { + // Warn-once is per continuous skip run, not per process. The suppression exists so a + // permanently portless row does not repeat every minute; carried across a repair it would + // silence the SECOND time a node loses its endpoint — the occurrence an operator most needs + // to see, because it is a regression rather than a known gap. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", null, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + SetGrpcPort(factory, "host-a:4053", 4056); + rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts()); + AwaitCondition( + () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5)); + + SetGrpcPort(factory, "host-a:4053", null); + EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(TimeSpan.FromSeconds(5), () => + rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts())); + } + [Fact] public void One_failing_node_does_not_stop_the_fan_and_is_logged_by_node_id() { @@ -690,29 +712,89 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe( new[] { "http://host-a:4056", "http://host-b:4056" }, ignoreOrder: true); + + // Once PER REFRESH, not once at startup. A key can rotate without any membership change, so + // an invalidation that only fired when the cluster set moved would serve the pre-rotation + // key forever and a rotation would need a central restart. + rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(2); } - [Fact] - public void The_known_cluster_set_is_not_published_by_the_other_two_modes() + [Theory] + [InlineData(MeshTransportOptions.ModeDps)] + [InlineData(MeshTransportOptions.ModeClusterClient)] + [InlineData(MeshTransportOptions.ModeGrpc)] + public void The_known_cluster_gate_is_published_in_every_transport_mode(string mode) { - // The gate is a gRPC-command-plane control. Publishing it under Dps or ClusterClient would - // start refusing acks on a transport whose ack path never consults the provider at all — - // and, worse, would do so from a code path those modes' tests do not cover. + // Replaces an earlier fact asserting the OPPOSITE, whose premise did not survive 1B: the + // gate feeds CentralCommandAuthInterceptor, which is registered on every admin node, and the + // ack service it protects is mapped by LISTENER PORT — neither consults MeshTransport:Mode. + // Gating the publication on Grpc meant a dark-switch rollback to ClusterClient or Dps left + // that endpoint serving with a permanently inactive gate, silently restoring both the timing + // oracle and the pre-authentication store/audit flood. The set is central's cluster + // inventory, not transport state. var factory = NewInMemoryDbFactory(); - SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false)); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); var psk = new RecordingPskProvider(); Sys.ActorOf(CentralCommunicationActor.Props( factory, - ClusterClientMode(), + new MeshTransportOptions { Mode = mode, ContactRefreshSeconds = 60 }, new RecordingClientFactory(() => CreateTestProbe()), () => null, - dialer: null, - pskProvider: psk)); + new RecordingDialer(), + psk)); - ExpectNoMsg(TimeSpan.FromMilliseconds(500)); - psk.KnownClusterPublications.ShouldBeEmpty(); - psk.InvalidateAllCount.ShouldBe(0); + AwaitCondition( + () => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5)); + psk.KnownClusterPublications[0].ShouldBe(new[] { "C1" }); + psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1); + } + + [Fact] + public void The_gate_set_includes_a_cluster_the_fan_set_excludes() + { + // The two sets answer different questions. The FAN asks "who does central talk to" — enabled, + // non-maintenance, ported. The GATE asks "is this a cluster central has ever heard of", and + // must say yes for a cluster whose every node is disabled or in maintenance: such a cluster + // can still send a deployment ack, and refusing it would quietly turn maintenance mode into + // an authentication switch. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + SeedGrpcCluster(factory, "C-DISABLED", ("host-b:4053", "host-b", 4056, false, false)); + SeedGrpcCluster(factory, "C-MAINT", ("host-c:4053", "host-c", 4056, true, true)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + rig.Psk.KnownClusterPublications[0].ShouldBe( + new[] { "C1", "C-DISABLED", "C-MAINT" }, ignoreOrder: true); + + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + rig.Dialer.Calls.Count.ShouldBe(1); + rig.Dialer.Calls[0].ClusterId.ShouldBe("C1"); + } + + [Fact] + public void The_gate_is_published_after_the_dispatch_set_is_adopted_and_before_the_cache_is_dropped() + { + // Two orderings in one handler, both load-bearing. SetKnownClusters must precede + // InvalidateAll, or the first lookup after a refresh re-reads the store under the PREVIOUS + // membership — briefly resolving a cluster central has just retired. And the dispatch set is + // adopted first so a command processed immediately after the publication already fans over + // the current targets. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + rig.Psk.Sequence.ShouldBe(new[] { "SetKnownClusters", "InvalidateAll" }); + + // Targets-before-gate, observed the only way it can be from outside the actor: the very next + // message finds the dispatch set already populated. + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); } [Fact] @@ -738,6 +820,15 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase () => actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()))); } + private static void SetGrpcPort( + IDbContextFactory factory, string nodeId, int? grpcPort) + { + using var db = factory.CreateDbContext(); + var node = db.ClusterNodes.Single(n => n.NodeId == nodeId); + node.GrpcPort = grpcPort; + db.SaveChanges(); + } + private static DispatchDeployment NewDispatch() => new( new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid())); @@ -839,8 +930,15 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase { private readonly object _gate = new(); private readonly List> _publications = []; + private readonly List _sequence = []; private int _invalidateAllCount; + /// Names of the mutating calls made, in the order they arrived. + public IReadOnlyList Sequence + { + get { lock (_gate) return _sequence.ToList(); } + } + /// Every known-cluster set published, in order. public IReadOnlyList> KnownClusterPublications { @@ -860,11 +958,19 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase public void SetKnownClusters(IReadOnlyCollection clusterIds) { - lock (_gate) _publications.Add(clusterIds.ToList()); + lock (_gate) + { + _publications.Add(clusterIds.ToList()); + _sequence.Add(nameof(SetKnownClusters)); + } } public void Invalidate(string clusterId) { } - public void InvalidateAll() => Interlocked.Increment(ref _invalidateAllCount); + public void InvalidateAll() + { + Interlocked.Increment(ref _invalidateAllCount); + lock (_gate) _sequence.Add(nameof(InvalidateAll)); + } } } 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 96a0db0f..be0896aa 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 @@ -1,6 +1,8 @@ using System.Diagnostics.CodeAnalysis; using Grpc.Core; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Shouldly; using Xunit; @@ -41,6 +43,7 @@ public sealed class MeshGrpcCommandDialerTests => new( psk, Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = dispatchTimeoutSeconds }), + NullLogger.Instance, _ => client); [Fact] @@ -136,17 +139,25 @@ public sealed class MeshGrpcCommandDialerTests } [Fact] - public async Task A_zero_dispatch_timeout_leaves_the_dial_bounded_only_by_the_callers_token() + public async Task A_non_positive_dispatch_timeout_is_clamped_rather_than_honoured() { - // The validator requires > 0 under Mode=Grpc, so this is the misconfigured/other-mode shape. - // It must not be read as "expire immediately" — that would drop every command instantly. - var client = new RecordingClient(); - var dialer = CreateDialer( - new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 0); + // Zero would mean an UNBOUNDED call — the exact failure the deadline exists to prevent — and + // MeshTransportOptionsValidator's rule for this key is scoped to driver-role nodes, so a + // pure-admin central reaches this code with the value unvalidated. Clamp loudly. + var client = new RecordingClient { HangUntilCancelled = true }; + var logger = new RecordingLogger(); + var dialer = new MeshGrpcCommandDialer( + new FakePskProvider(("SITE-A", "psk-site-a")), + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 0 }), + logger, + _ => client); - await dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None); + logger.Warnings.ShouldContain(w => w.Contains("DispatchTimeoutSeconds")); - client.Calls.ShouldBe(1); + // And the clamp is real, not just logged: the dial must still expire on its own. + await Should.ThrowAsync( + () => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None) + .WaitAsync(TimeSpan.FromSeconds(20))); } [Fact] @@ -169,7 +180,8 @@ public sealed class MeshGrpcCommandDialerTests // and Dispose must then clean up the channel the dial created. var dialer = new MeshGrpcCommandDialer( new FakePskProvider(("SITE-A", "psk-site-a")), - Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 5 })); + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 5 }), + NullLogger.Instance); await Should.ThrowAsync( () => dialer.DispatchAsync( @@ -178,6 +190,38 @@ public sealed class MeshGrpcCommandDialerTests Should.NotThrow(dialer.Dispose); } + [Fact] + public async Task Two_dials_to_one_endpoint_share_a_single_channel() + { + // The cache is the whole defence against a socket per call: a deploy dispatch fans across + // every node in the fleet, and it repeats on every redeploy. A channel per dial exhausts + // ports on central long before anything else notices. + var dialer = new MeshGrpcCommandDialer( + new FakePskProvider(("SITE-A", "psk-site-a")), + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 5 }), + NullLogger.Instance); + + // Nothing is listening, so both dials fail — irrelevant here: the channel is constructed + // before the call is placed, which is exactly what is being counted. + for (var i = 0; i < 2; i++) + { + await Should.ThrowAsync( + () => dialer.DispatchAsync( + "http://127.0.0.1:1", Envelope(), "SITE-A", CancellationToken.None)); + } + + dialer.CachedChannelCount.ShouldBe(1); + + // A second endpoint is a second channel — the cache keys on the endpoint, it does not + // collapse every node onto one connection. + await Should.ThrowAsync( + () => dialer.DispatchAsync( + "http://127.0.0.1:2", Envelope(), "SITE-A", CancellationToken.None)); + + dialer.CachedChannelCount.ShouldBe(2); + Should.NotThrow(dialer.Dispose); + } + [Fact] public void It_activates_from_the_container_despite_the_optional_client_factory_parameter() { @@ -186,6 +230,7 @@ public sealed class MeshGrpcCommandDialerTests // constructor — and the failure is a startup crash on the admin node, invisible to every // other test here because they all construct the dialer by hand. var services = new ServiceCollection(); + services.AddLogging(); services.AddSingleton(new FakePskProvider(("SITE-A", "psk-site-a"))); services.AddSingleton>( Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 10 })); @@ -249,6 +294,28 @@ public sealed class MeshGrpcCommandDialerTests } } + /// Captures Warning-level messages so a construction-time check is assertable. + private sealed class RecordingLogger : ILogger + { + /// Every warning logged, formatted. + public List Warnings { get; } = []; + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + if (logLevel == LogLevel.Warning) + Warnings.Add(formatter(state, exception)); + } + } + /// An over a fixed key set; unknown clusters fail closed. private sealed class FakePskProvider(params (string Cluster, string Psk)[] keys) : IMeshPskProvider { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs index bb619709..510a5dbc 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs @@ -248,8 +248,8 @@ public sealed class CentralCommandAuthInterceptorTests // Phase 1A security register item 2. The interceptor must not fetch the raw key and compare // it — that is how a constant-time comparison eventually becomes '=='. Matches owns the // comparison; TryGetPsk survives only for the DIAL side, which has to send the key. - // (The provider's own default Matches calls TryGetPsk internally; this fake overrides it, so - // a zero TryGetPsk count here means the INTERCEPTOR never asked for the key.) + // (The production provider's Matches calls TryGetPsk internally; this fake implements the two + // independently, so a zero TryGetPsk count here means the INTERCEPTOR never asked for the key.) var provider = new FakeMeshPskProvider(("SITE-A", "psk-site-a")); var interceptor = new CentralCommandAuthInterceptor( provider, NullLogger.Instance); @@ -279,8 +279,10 @@ public sealed class CentralCommandAuthInterceptorTests } /// - /// Overrides the interface's default so the two calls are separately countable. The - /// semantics are the default's: no key, or a key that is not exactly the presented token. + /// Implemented independently of so the two calls are separately + /// countable — the interface declares this with no default implementation, deliberately. + /// Semantics match the production provider's: no key, or a key that is not exactly the + /// presented token. /// public bool Matches(string clusterId, string presented) {