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) {