From 6a86201105a9ef17ba0d3efaf8fafc5479a41348 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 05:58:21 -0400 Subject: [PATCH] feat(mesh): gRPC command fan-out behind Mode=Grpc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MeshGrpcCommandDialer places one NodeCommandService.Execute call on one node: a GrpcChannel cached per endpoint (the GrpcDeploymentArtifactFetcher shape), the TARGET node's cluster's bearer plus x-otopcua-cluster naming that cluster, and a per-call deadline linked to the caller's token. A cluster with no resolvable key is refused locally with Unauthenticated — there is no arm on which a command leaves central unauthenticated, even one the far side would have rejected. CentralCommunicationActor gains the third mode branch. The contact load now also projects GrpcPort and, from the SAME rows in the SAME pass, builds the dispatch set — which is what makes 'the gRPC fan reaches exactly the SendToAll recipient set' structural rather than two queries that have to keep agreeing. A row with a null GrpcPort is skipped warn-once (the TelemetryNodeSource precedent). Per-node results are piped back for LOGGING ONLY: named node, gRPC status, dropped. No retry, no queue. Every dial is scheduled on the thread pool rather than called from the receive. Found by review and pinned red-first: an async method runs synchronously to its first await, and the dialer's pre-await work is TryGetPsk, which blocks on the secret store — bounded but seconds, and repeated after every refresh since the refresh drops the cache. Dialing inline parked the actor's mailbox, so deployment acks queued behind a slow node; the regression test tells an ApplyAck while a dial is held and fails without the hop. The contact refresh also wires the Phase 1A register: it publishes the loaded cluster ids as the PSK provider's known-cluster set and drops the key cache, so a rotated key takes effect within one ContactRefreshSeconds. Neither happens under Dps or ClusterClient. Both existing mode branches are behaviourally untouched and every existing fact passes unmodified. The one edit to shared text: the Status.Failure warning drops the word 'ClusterClient', since that handler now serves all three transports. ControlPlane.Tests 205 -> 230, Host.Tests 92, Cluster.Tests 184, 0 warnings. --- .../CentralCommunicationActor.cs | 427 ++++++++++++- .../Communication/MeshGrpcCommandDialer.cs | 159 +++++ .../ServiceCollectionExtensions.cs | 13 +- src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 8 + .../CentralCommunicationActorTests.cs | 561 ++++++++++++++++++ .../MeshGrpcCommandDialerTests.cs | 271 +++++++++ 6 files changed, 1428 insertions(+), 11 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs 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 0c470ede..635194ff 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs @@ -3,17 +3,24 @@ using Akka.Actor; using Akka.Cluster.Tools.Client; using Akka.Cluster.Tools.PublishSubscribe; using Akka.Event; +using Grpc.Core; using Microsoft.EntityFrameworkCore; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; +// Grpc.Core (imported for RpcException on the dial path) declares a Status of its own. This file's +// bare Status is Akka's — the Status.Failure a faulted piped task arrives as; the gRPC one is only +// ever reached through an RpcException instance. +using Status = Akka.Actor.Status; namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; /// /// Central-side end of the central↔node command boundary (per-cluster mesh Phase 2). Routes -/// every central→node command out — over DistributedPubSub or ClusterClient, per +/// every central→node command out — over DistributedPubSub, ClusterClient or gRPC, per /// — and forwards inbound s to the /// deploy coordinator singleton. /// @@ -31,6 +38,23 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; /// and fans SendToAll across all of them. See — this is the /// single most important thing to understand before changing this actor. /// +/// +/// The gRPC branch fans over the same recipient set, node by node +/// (ClusterClient→gRPC Phase 1A). SendToAll reaches every node registered on a mesh's +/// receptionist, and the contact load already selects exactly the enabled, non-maintenance +/// rows those nodes come from; the gRPC branch dials that same row set directly, one call per +/// node, so the recipient set is identical by construction rather than by coincidence. It is +/// deliberately NOT narrower: the node side has no ClusterId or node filter at all — +/// scoping happens later, inside the artifact — so a node left out here silently keeps its +/// old configuration while the deployment seals green. +/// +/// +/// Fire-and-forget, on every transport. The gRPC branch pipes per-node results back to +/// itself for logging ONLY: a failure names the node and the status and is dropped. No retry, +/// no queue, no ordering guarantee across nodes. That is not an omission — it is the +/// ClusterClient semantic preserved exactly, and the backstops are unchanged (the deploy +/// coordinator's apply deadline, and a node's boot-time replay). +/// /// public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers { @@ -40,8 +64,11 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers private readonly MeshTransportOptions _options; private readonly IMeshClusterClientFactory _clientFactory; private readonly Func _coordinator; + private readonly IMeshCommandDialer? _dialer; + private readonly IMeshPskProvider? _pskProvider; private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly bool _useClusterClient; + private readonly bool _useGrpc; // One client per application Cluster, keyed by ClusterId. A ClusterClient reaches only the mesh // whose receptionist it dialled, and after Phase 6 every cluster is its own mesh — so central must @@ -49,6 +76,15 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers private readonly Dictionary Contacts)> _clients = new(); + // Node ids already warned about for a null GrpcPort. A persistently portless row would otherwise + // produce one warning per refresh forever; same warn-once shape as TelemetryNodeSource. Actor + // thread only. + private readonly HashSet _warnedPortlessNodes = new(StringComparer.Ordinal); + + // The gRPC dispatch set, replaced wholesale on every contact load. Actor thread only; the fan + // captures the list reference, so a refresh mid-fan cannot mutate a dispatch already under way. + private IReadOnlyList _targets = []; + private CancellationTokenSource _lifecycle = new(); /// Gets the timer scheduler driving the periodic contact refresh. @@ -63,46 +99,73 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// an so this actor never depends on registration order at wiring /// time — it resolves on first use and caches. /// + /// + /// Places one command call on one node, under . + /// Optional so the two older transports need nothing new wired; a + /// dialer under Grpc is a mis-wiring and is reported loudly rather than silently + /// dropping every command. + /// + /// + /// Central's per-cluster key lookup, told on every contact refresh which clusters exist (the + /// known-cluster gate) and to drop its cache (so a rotated key takes effect within one + /// refresh). Optional for the same reason as . + /// /// The props. public static Props Props( IDbContextFactory dbFactory, MeshTransportOptions options, IMeshClusterClientFactory clientFactory, - Func coordinator) => + Func coordinator, + IMeshCommandDialer? dialer = null, + IMeshPskProvider? pskProvider = null) => Akka.Actor.Props.Create(() => - new CentralCommunicationActor(dbFactory, options, clientFactory, coordinator)); + new CentralCommunicationActor( + dbFactory, options, clientFactory, coordinator, dialer, pskProvider)); /// Initializes a new instance of the class. /// Factory for the config database. /// The bound mesh-transport options. /// Creates the ClusterClient. /// Lazily resolves the deploy-coordinator singleton proxy. + /// Places one gRPC command call on one node; Grpc mode only. + /// Central's per-cluster key lookup; Grpc mode only. public CentralCommunicationActor( IDbContextFactory dbFactory, MeshTransportOptions options, IMeshClusterClientFactory clientFactory, - Func coordinator) + Func coordinator, + IMeshCommandDialer? dialer = null, + IMeshPskProvider? pskProvider = null) { _dbFactory = dbFactory; _options = options; _clientFactory = clientFactory; _coordinator = coordinator; + _dialer = dialer; + _pskProvider = pskProvider; _useClusterClient = string.Equals( options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); + _useGrpc = string.Equals( + options.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); Receive(HandleMeshCommand); Receive(HandleApplyAck); Receive(_ => LoadContactsFromDb()); Receive(HandleContactsLoaded); + Receive(HandleCommandDispatched); // A faulted LoadContactsFromDb task is piped here as a Status.Failure. Without this handler // the failure is an unhandled message (debug level only) and the refresh fails silently — // 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 ClusterClient contact set was NOT " - + "refreshed and may be stale or empty. Commands may be dropped until the next refresh")); + "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")); Receive(_ => { /* DPS subscribe confirmation */ }); } @@ -110,7 +173,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// protected override void PreStart() { - if (!_useClusterClient) + if (!_useClusterClient && !_useGrpc) { _log.Info( "Mesh transport is {Mode}: commands are published on DistributedPubSub and no " @@ -118,6 +181,19 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers return; } + if (_useGrpc && _dialer is null) + { + // Not a fatal throw: the ack path (HandleApplyAck) is transport-independent and still + // works, so taking the actor down would turn a command-plane mis-wiring into an ack + // outage as well. Every dispatch logs its own error, so this cannot go unnoticed. + _log.Error( + "Mesh transport is Grpc but no command dialer was supplied to this actor; EVERY " + + "central→node command will be dropped. Check the ControlPlane registration wires " + + "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. Timers.StartPeriodicTimer( RefreshTimerKey, new RefreshContacts(), @@ -131,12 +207,23 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers foreach (var (client, _) in _clients.Values) Context.Stop(client); _clients.Clear(); + // Cancelling the lifecycle token also cancels every gRPC dial in flight. The dialer itself is + // deliberately NOT disposed here: it is a container-owned singleton shared with anything else + // that may dial, and an actor restart must not tear its channel pool down. _lifecycle.Cancel(); _lifecycle.Dispose(); } private void HandleMeshCommand(MeshCommand cmd) { + // Checked before the DPS arm: _useClusterClient is false under Grpc too, so the order is + // what keeps a Grpc-mode command off DistributedPubSub. + if (_useGrpc) + { + DispatchOverGrpc(cmd); + return; + } + if (!_useClusterClient) { DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, cmd.Message)); @@ -168,6 +255,199 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, cmd.Message)); } + /// + /// Encodes one command once and fans it to every target node, one unary gRPC call each. + /// + /// The command envelope carrying the domain record. + /// + /// + /// Encoded once, before the fan. Not per node — partly because the wire form is + /// identical for every recipient, but mostly because AlarmCommand has no + /// correlation member of its own and the envelope mints one: minting inside the loop + /// would give one operator action N unrelated correlation ids and make it untraceable. + /// + /// + /// Nothing captured from the actor. Each dial is a static call taking only locals, + /// so no per-node continuation can touch actor state off the actor's thread. Results come + /// back as ordinary messages via PipeTo and are handled on the actor thread like + /// anything else. + /// + /// + /// Every dial starts on the thread pool, not here. An async method runs + /// synchronously up to its first await, and the dialer's pre-await work includes + /// IMeshPskProvider.TryGetPsk, which blocks on the secret store (bounded, but + /// seconds) the first time a cluster is looked up — and again after every refresh, since + /// the refresh drops the cache to pick up rotations. Calling straight into it would park + /// this actor's mailbox, and deployment acks queue behind that mailbox. + /// + /// + private void DispatchOverGrpc(MeshCommand cmd) + { + var commandType = cmd.Message.GetType().Name; + + var dialer = _dialer; + if (dialer is null) + { + _log.Error( + "Mesh transport is Grpc but no command dialer is wired — dropping {MessageType}. The " + + "command is not buffered and will not be retried", + commandType); + return; + } + + // Captured before the fan: a contact refresh between here and the last dial must not change + // the recipient set of a dispatch already under way. + var targets = _targets; + if (targets.Count == 0) + { + // App-level drop, exactly as the ClusterClient arm does with no client. The command is + // NOT queued and will NOT be retried: a deploy fails at the coordinator's apply deadline + // naming the silent nodes, which is a better operator signal than a command arriving + // minutes late against a config the DB already recorded as failed. + _log.Warning( + "No mesh gRPC targets — dropping {MessageType}. The last contact refresh produced no " + + "enabled, non-maintenance ClusterNode row carrying a GrpcPort in any cluster; the " + + "command is not buffered and will not be retried", + commandType); + return; + } + + CommandEnvelope envelope; + try + { + envelope = MeshTransportDtoMapper.ToEnvelope(cmd.Message, new CorrelationId(Guid.NewGuid())); + } + catch (Exception ex) + { + // The mapper refuses a payload the far side could not decode. Dropping here is the same + // outcome the old transport gave for such a message (nothing on the node matched it), but + // loud instead of silent. + _log.Error( + ex, + "Cannot encode {MessageType} for the gRPC command plane; the command is dropped and " + + "was NOT delivered to any node", + commandType); + return; + } + + CancellationToken ct; + try + { + ct = _lifecycle.Token; + } + catch (ObjectDisposedException) + { + return; // Stopping. + } + + var self = Self; + var correlationId = envelope.CorrelationId; + foreach (var target in targets) + { + var node = target; + + // CancellationToken.None on Task.Run, deliberately: cancelling the SCHEDULING would make + // the task complete as canceled, which PipeTo converts into a Status.Failure — the one + // message this actor already reads as "the contact load failed". The lifecycle token is + // passed into the dial instead, where it cancels the call and comes back as an ordinary + // per-node outcome. + Task.Run( + () => DispatchOneAsync(dialer, node, envelope, commandType, correlationId, ct), + CancellationToken.None) + .PipeTo(self); + } + } + + /// + /// Dials one node and turns every possible outcome into a . + /// + /// The dialer seam. + /// The node to dial. + /// The already-encoded command. + /// CLR type name of the command, for the log. + /// Transport correlation, for the log. + /// The actor's lifecycle token. + /// The outcome, always — this method does not fault. + /// + /// Catching everything is what keeps the result on ONE channel. A faulting task piped to an + /// actor arrives as Status.Failure, which this actor already handles as "the contact + /// load failed" — so an unhandled dial fault would log a database-outage warning about a node + /// that simply did not answer. + /// + private static async Task DispatchOneAsync( + IMeshCommandDialer dialer, + MeshNodeTarget target, + CommandEnvelope envelope, + string commandType, + string correlationId, + CancellationToken ct) + { + try + { + await dialer.DispatchAsync(target.Endpoint, envelope, target.ClusterId, ct) + .ConfigureAwait(false); + return new CommandDispatched(target.NodeId, commandType, correlationId, null, null); + } + catch (RpcException ex) + { + return new CommandDispatched( + target.NodeId, + commandType, + correlationId, + ex.StatusCode.ToString(), + Truncate(ex.Status.Detail)); + } + catch (OperationCanceledException) + { + // Named rather than swept into the generic arm: this is the ordinary shape of a node that + // did not answer in time, and "Unknown / TaskCanceledException" would send an operator + // looking for a bug instead of at DispatchTimeoutSeconds. + return new CommandDispatched( + target.NodeId, + commandType, + correlationId, + "Cancelled", + "the dial did not complete within MeshTransport:DispatchTimeoutSeconds, or central " + + "is shutting down"); + } + catch (Exception ex) + { + // Type name only, never the message: this path also covers whatever the dialer's channel + // construction can throw, and a message is the one place a misconfigured endpoint string + // could carry something an operator did not mean to put in a log. + return new CommandDispatched( + target.NodeId, commandType, correlationId, "Unknown", ex.GetType().Name); + } + } + + /// + /// Bounds a status detail before it reaches the log. The string comes from the far side, and + /// the far side is a node — trusted, but not a reason to let it choose how many lines of + /// central's log one failed command occupies. + /// + /// The received status detail. + /// The detail, at most 200 characters. + private static string Truncate(string? detail) + => detail is null || detail.Length <= 200 ? detail ?? "" : detail[..200] + "…"; + + private void HandleCommandDispatched(CommandDispatched result) + { + if (result.FailureStatus is null) + { + _log.Debug( + "Mesh: {CommandType} (correlationId={CorrelationId}) accepted by {NodeId} over gRPC", + result.CommandType, result.CorrelationId, result.NodeId); + return; + } + + _log.Warning( + "Mesh: {CommandType} (correlationId={CorrelationId}) was NOT delivered to {NodeId} " + + "({Status}: {Detail}). It is dropped and will NOT be retried — the deployment apply " + + "deadline and the node's boot-time replay are the backstops", + result.CommandType, result.CorrelationId, result.NodeId, + result.FailureStatus, result.FailureDetail); + } + private void HandleApplyAck(ApplyAck ack) { var coordinator = _coordinator(); @@ -210,7 +490,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers var rows = await db.ClusterNodes .AsNoTracking() .Where(n => n.Enabled && !n.MaintenanceMode) - .Select(n => new { n.NodeId, n.ClusterId, n.Host, n.AkkaPort }) + .Select(n => new { n.NodeId, n.ClusterId, n.Host, n.AkkaPort, n.GrpcPort }) .ToListAsync(ct) .ConfigureAwait(false); @@ -219,8 +499,28 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers // client. var byCluster = new Dictionary>(); var malformed = new List(); + + // The gRPC dispatch set, built from the SAME row set in the SAME pass. Sharing the read is + // what makes "the gRPC fan reaches exactly the ClusterClient recipient set" structural + // rather than a claim two separate queries would have to keep agreeing on. + 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. + var clusterIds = new HashSet(StringComparer.OrdinalIgnoreCase); + foreach (var row in rows) { + clusterIds.Add(row.ClusterId); + + if (row.GrpcPort is { } grpcPort) + targets.Add(new MeshNodeTarget( + row.NodeId, row.ClusterId, $"http://{row.Host}:{grpcPort}")); + else + portless.Add(row.NodeId); + var address = $"akka.tcp://{systemName}@{row.Host}:{row.AkkaPort}/system/receptionist"; // Parse up front, per row, inside the loop's own guard. A single malformed row must @@ -241,12 +541,19 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers var contactsByCluster = byCluster.ToDictionary( kv => kv.Key, kv => (IReadOnlyList)kv.Value); - return new ContactsLoaded(contactsByCluster, malformed); + return new ContactsLoaded( + contactsByCluster, malformed, targets, portless, clusterIds.ToList()); }, ct).PipeTo(self); } private void HandleContactsLoaded(ContactsLoaded msg) { + if (_useGrpc) + { + HandleGrpcContactsLoaded(msg); + return; + } + foreach (var bad in msg.Malformed) { _log.Warning( @@ -305,6 +612,59 @@ 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. + /// + /// 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. + /// + /// + private void HandleGrpcContactsLoaded(ContactsLoaded msg) + { + 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. + if (_warnedPortlessNodes.Add(nodeId)) + { + _log.Warning( + "ClusterNode {NodeId} has no GrpcPort; it exposes no command endpoint and is " + + "skipped in this dispatch set (further skips of this node are silent). Commands " + + "will not reach it until the row carries a port", nodeId); + } + } + + _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( + "No enabled, non-maintenance ClusterNode row carries a GrpcPort in any cluster; there " + + "are no gRPC command targets and every central→node command will be dropped until a " + + "refresh finds one"); + } + } + /// Stops a cluster's existing client and creates a replacement for its new contact set. /// /// @@ -371,7 +731,54 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// Cluster's id. One ClusterClient is built per key. /// /// Rows that did not parse, described for the log. + /// + /// The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a + /// GrpcPort. Built from the same rows as . + /// + /// + /// Node ids skipped because their row has no GrpcPort. Carried rather than logged in + /// 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. + /// public sealed record ContactsLoaded( IReadOnlyDictionary> ContactsByCluster, - IReadOnlyList Malformed); + IReadOnlyList Malformed, + IReadOnlyList Targets, + IReadOnlyList PortlessNodes, + IReadOnlyList ClusterIds); + + /// One node the gRPC command plane can dial. + /// The ClusterNode row's id — what a failure log names. + /// + /// The node's application Cluster. Selects the pre-shared key presented to it: a node + /// dialled with another cluster's id is graded against the wrong key and refuses the command. + /// + /// The h2c base address, http://{Host}:{GrpcPort}. + public sealed record MeshNodeTarget(string NodeId, string ClusterId, string Endpoint); + + /// + /// The outcome of one node's dial, piped back to the actor for logging only. Nothing + /// retries, counts or waits on these — a deployment's completion is judged by the acks nodes + /// send back, never by whether central's outbound call succeeded. + /// + /// The node dialled. + /// CLR type name of the command. + /// The envelope's transport correlation. + /// + /// on success; otherwise the gRPC status code's name, or + /// Unknown for a non-gRPC fault. + /// + /// + /// The gRPC status detail, or the exception's type name for a non-gRPC fault. Never a key, + /// and never a raw exception message. + /// + public sealed record CommandDispatched( + string NodeId, + string CommandType, + string CorrelationId, + string? FailureStatus, + string? FailureDetail); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs new file mode 100644 index 00000000..46dcd830 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs @@ -0,0 +1,159 @@ +using System.Collections.Concurrent; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; + +/// +/// Places one central→node command call on one node's NodeCommandService +/// (ClusterClient→gRPC migration Phase 1A). The seam CentralCommunicationActor fans +/// across its target set, kept as an interface so the actor's tests never open a socket. +/// +/// +/// One dial, one node, no retry and no fan-out of its own — the fan and its per-node failure +/// logging belong to the actor, which is the component that knows what the recipient set is. +/// +public interface IMeshCommandDialer +{ + /// Delivers one command envelope to one node. + /// The node's gRPC base address, e.g. http://site-a-1:4056. + /// The wire envelope, already mapped from the domain record. + /// + /// The application Cluster the target node belongs to. Selects which pre-shared + /// key is presented, and is echoed to the node in x-otopcua-cluster. + /// + /// Cancels the dial; a per-call deadline is applied on top of it. + /// A task completing when the node has accepted the command. + /// + /// The node rejected the call, was unreachable, or the deadline expired — and also when no + /// key could be established for , which is refused locally with + /// rather than dialled unauthenticated. + /// + Task DispatchAsync(string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct); +} + +/// +/// +/// +/// Fail-closed on a missing key, always. There is no arm on which a command leaves +/// central without the target cluster's bearer. A node's MeshCommandAuthInterceptor +/// would reject an unauthenticated call anyway, but "the far side would have refused it" is +/// not a reason to make the call — an unauthenticated dial is a credential-free command +/// attempt against a node's control surface, and it should not exist even as a rejected one. +/// The refusal is reported as an so the actor's per-node failure +/// logging covers it on the same channel as a remote rejection; the operator response is +/// identical either way (seed the cluster's key), and a second channel would need a second +/// handler that could drift. +/// +/// +/// One channel per endpoint, cached. Same shape as +/// GrpcDeploymentArtifactFetcher: a owns a connection pool +/// and creating one per call leaks sockets. The cache is keyed by the endpoint string +/// ordinally — endpoints come from ClusterNode.Host + GrpcPort, so two spellings +/// of one host would merely cost a second channel, never a wrong destination. +/// +/// +/// The deadline is per call, not per fan. A dispatch dials every node in the fleet; +/// an unbounded call to one unreachable node would hold a task and a socket indefinitely +/// while the deploy path's own 2-minute apply deadline ticked down. +/// bounds it, linked to the +/// 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. +/// +/// +public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable +{ + /// + /// 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 + /// a node which key central graded itself against. + /// + private const string ClusterHeader = "x-otopcua-cluster"; + + private readonly IMeshPskProvider _pskProvider; + private readonly int _dispatchTimeoutSeconds; + private readonly Func _clientFactory; + private readonly ConcurrentDictionary _channels = new(StringComparer.Ordinal); + + private volatile bool _disposed; + + /// Initializes a new instance of the class. + /// Supplies the bearer for the target node's cluster. + /// Mesh transport options; supplies the per-call deadline. + /// + /// Seam for tests: builds the generated client for an endpoint. Defaults to a real h2c + /// , one cached per endpoint. + /// + public MeshGrpcCommandDialer( + IMeshPskProvider pskProvider, + IOptions options, + Func? clientFactory = null) + { + ArgumentNullException.ThrowIfNull(pskProvider); + ArgumentNullException.ThrowIfNull(options); + + _pskProvider = pskProvider; + _dispatchTimeoutSeconds = options.Value.DispatchTimeoutSeconds; + _clientFactory = clientFactory ?? DefaultClientFactory; + } + + /// + public async Task DispatchAsync( + string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct) + { + ArgumentException.ThrowIfNullOrWhiteSpace(nodeEndpoint); + ArgumentNullException.ThrowIfNull(envelope); + ArgumentException.ThrowIfNullOrWhiteSpace(clusterId); + ObjectDisposedException.ThrowIf(_disposed, this); + + if (!_pskProvider.TryGetPsk(clusterId, out var psk)) + { + // 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. + 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.")); + } + + var headers = new Metadata + { + { "authorization", $"Bearer {psk}" }, + { ClusterHeader, clusterId }, + }; + + using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + if (_dispatchTimeoutSeconds > 0) + deadlineCts.CancelAfter(TimeSpan.FromSeconds(_dispatchTimeoutSeconds)); + + var client = _clientFactory(nodeEndpoint); + using var call = client.ExecuteAsync(envelope, headers, cancellationToken: deadlineCts.Token); + await call.ResponseAsync.ConfigureAwait(false); + } + + /// + public void Dispose() + { + _disposed = true; + + foreach (var channel in _channels.Values) + channel.Dispose(); + _channels.Clear(); + } + + private NodeCommandService.NodeCommandServiceClient DefaultClientFactory(string endpoint) + { + // h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. + var channel = _channels.GetOrAdd(endpoint, GrpcChannel.ForAddress); + return new NodeCommandService.NodeCommandServiceClient(channel); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 08cf21e9..32a04231 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -66,6 +66,15 @@ public static class ServiceCollectionExtensions var dbFactory = resolver.GetService>(); var options = resolver.GetService>().Value; + // Through IServiceProvider rather than the Akka resolver, deliberately: these two are + // OPTIONAL (Grpc mode only, and admin-role only within that), and GetService's + // null-for-absent contract is the thing being relied on. The Akka resolver's + // GetService makes no such promise, and this same block runs in test harnesses whose + // service collections register neither. + var serviceProvider = resolver.GetService(); + var dialer = serviceProvider.GetService(); + var pskProvider = serviceProvider.GetService(); + var actor = system.ActorOf( CentralCommunicationActor.Props( dbFactory, @@ -75,7 +84,9 @@ public static class ServiceCollectionExtensions // which Akka.Hosting materialises the singleton proxy relative to this block. () => registry.TryGet(out var coordinator) ? coordinator - : null), + : null, + dialer, + pskProvider), MeshPaths.CentralCommunicationName); // Registered unconditionally, in both transport modes. Under Dps nothing dials it, and diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 33ff5a43..2fb407a3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -534,6 +534,14 @@ if (hasAdmin) // Depends on ISecretResolver, which AddOtOpcUaSecrets registers unconditionally below (DI // registration order is irrelevant; resolution order is not). builder.Services.AddSingleton(); + // Central's outbound command dial (ClusterClient→gRPC Phase 1A). Admin-only for the same reason + // as the provider above — a node never dials another node — and singleton because it owns a + // GrpcChannel per node endpoint; per-call channels would leak sockets across a fleet-wide fan. + // Registered in every mode: CentralCommunicationActor only reaches it under Mode=Grpc, and an + // idle registration costs nothing, whereas gating it here would mean flipping the mode on a + // running fleet needs a central restart before any command lands — the same reasoning that + // registers the ClusterClient receptionist unconditionally. + builder.Services.AddSingleton(); } // Registered unconditionally: driver-role nodes resolve Layer-B DriverConfig secrets and have no auth/DP/AdminUI. 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 97e3f28f..8563f781 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 @@ -1,4 +1,5 @@ using System.Collections.Immutable; +using System.Diagnostics.CodeAnalysis; using Akka.Actor; using Akka.Cluster.Tools.Client; using Akka.Cluster.Tools.PublishSubscribe; @@ -9,6 +10,8 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; @@ -56,6 +59,13 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase ContactRefreshSeconds = refreshSeconds, }; + private static MeshTransportOptions GrpcMode(int refreshSeconds = 60) => new() + { + Mode = MeshTransportOptions.ModeGrpc, + ContactRefreshSeconds = refreshSeconds, + DispatchTimeoutSeconds = 10, + }; + private static void SeedCluster( IDbContextFactory factory, string clusterId, @@ -92,6 +102,46 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase db.SaveChanges(); } + /// + /// Seeds a cluster whose nodes carry a GrpcPort — the gRPC command plane's shape. + /// Kept separate from so the ClusterClient facts, which have no + /// opinion about the port, stay exactly as they were. + /// + private static void SeedGrpcCluster( + IDbContextFactory factory, + string clusterId, + params (string NodeId, string Host, int? GrpcPort, bool Enabled, bool Maintenance)[] nodes) + { + using var db = factory.CreateDbContext(); + db.ServerClusters.Add(new ServerCluster + { + ClusterId = clusterId, + Name = clusterId, + Enterprise = "ZB", + Site = "Site", + RedundancyMode = RedundancyMode.None, + NodeCount = 1, + CreatedBy = "test", + }); + foreach (var n in nodes) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = n.NodeId, + ClusterId = clusterId, + Host = n.Host, + AkkaPort = 4053, + GrpcPort = n.GrpcPort, + ApplicationUri = $"urn:{n.NodeId}", + Enabled = n.Enabled, + MaintenanceMode = n.Maintenance, + CreatedBy = "test", + }); + } + + db.SaveChanges(); + } + private IActorRef Spawn( IDbContextFactory factory, MeshTransportOptions options, @@ -306,4 +356,515 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase actor.Tell(new MeshCommand(DeploymentsTopic, "anything")); ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } + + // ------------------------------------------------------------------ Mode = Grpc + // + // The third transport. The fan-out must reach EXACTLY the recipient set SendToAll reaches today + // — every enabled, non-maintenance node in every cluster — because the node side still has no + // ClusterId or node filter; scoping happens later, inside the artifact. A narrower fan would + // leave nodes silently on their old configuration while the deployment sealed green. + + [Fact] + public void Grpc_mode_fans_one_command_to_every_enabled_node_across_all_clusters() + { + // THE assertion of this task, and the direct counterpart of + // A_command_fans_out_a_send_to_all_to_every_cluster_client. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster( + factory, + "C1", + ("host-a:4053", "host-a", 4056, true, false), + ("host-b:4053", "host-b", 4056, true, false)); + SeedGrpcCluster(factory, "C2", ("host-c:4053", "host-c", 4056, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + AwaitCondition(() => rig.Dialer.Calls.Count == 3, TimeSpan.FromSeconds(5)); + rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe( + new[] { "http://host-a:4056", "http://host-b:4056", "http://host-c:4056" }, ignoreOrder: true); + + // Each node is dialled with ITS OWN cluster's id — that is what selects the pre-shared key, + // so getting it wrong turns every command into a PermissionDenied at the node. + rig.Dialer.Calls.Single(c => c.Endpoint.Contains("host-c")).ClusterId.ShouldBe("C2"); + rig.Dialer.Calls.Where(c => !c.Endpoint.Contains("host-c")).ShouldAllBe(c => c.ClusterId == "C1"); + } + + [Fact] + public void Grpc_mode_creates_no_ClusterClient_and_publishes_nothing_on_pubsub() + { + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var clients = new RecordingClientFactory(() => CreateTestProbe()); + var mediatorProbe = CreateTestProbe(); + DistributedPubSub.Get(Sys).Mediator.Tell( + new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref); + mediatorProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + + var rig = SpawnGrpcRig(factory, clientFactory: clients); + AwaitContactsLoaded(rig); + + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); + clients.Calls.ShouldBeEmpty(); + mediatorProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + [Fact] + public void Grpc_mode_does_not_dial_disabled_or_maintenance_nodes() + { + // Same recipient filter as the ClusterClient contact load, asserted independently so the two + // transports cannot drift on who counts as a node. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster( + factory, + "C1", + ("host-a:4053", "host-a", 4056, true, false), + ("host-b:4053", "host-b", 4056, false, false), + ("host-c:4053", "host-c", 4056, true, true)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); + rig.Dialer.Calls[0].Endpoint.ShouldContain("host-a"); + + // And no late arrival from the two excluded rows. + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + rig.Dialer.Calls.Count.ShouldBe(1); + } + + [Fact] + public void A_node_with_no_GrpcPort_is_skipped_with_a_warning_and_the_others_are_still_dialled() + { + // Mirrors TelemetryNodeSource: a row with no port exposes no endpoint yet. It must cost that + // node its commands, never the whole fan. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster( + factory, + "C1", + ("host-a:4053", "host-a", null, true, false), + ("host-b:4053", "host-b", 4056, true, false)); + + GrpcRig? rig = null; + EventFilter.Warning(contains: "has no GrpcPort").ExpectOne( + TimeSpan.FromSeconds(5), () => rig = SpawnGrpcRig(factory)); + + rig.ShouldNotBeNull(); + AwaitContactsLoaded(rig); + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); + rig.Dialer.Calls[0].Endpoint.ShouldBe("http://host-b:4056"); + } + + [Fact] + public void One_failing_node_does_not_stop_the_fan_and_is_logged_by_node_id() + { + // Fire-and-forget: a failure is logged and DROPPED. No retry, no queue, and above all no + // abort of the remaining nodes — the deploy path's apply deadline is the backstop. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster( + factory, + "C1", + ("host-a:4053", "host-a", 4056, true, false), + ("host-b:4053", "host-b", 4056, true, false)); + var dialer = new RecordingDialer + { + FailEndpointsContaining = "host-a", + Fault = new Grpc.Core.RpcException( + new Grpc.Core.Status(Grpc.Core.StatusCode.Unavailable, "node down")), + }; + var rig = SpawnGrpcRig(factory, dialer); + AwaitContactsLoaded(rig); + + // The node id, not the endpoint: an operator reading the log is looking at ClusterNode rows. + EventFilter.Warning(contains: "host-a:4053").ExpectOne(TimeSpan.FromSeconds(5), () => + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()))); + + // Both were attempted; the healthy one is unaffected by its neighbour's failure. + AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); + + // Still serving after a per-node failure — no supervision restart, no state loss. + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + AwaitCondition(() => rig.Dialer.Calls.Count == 4, TimeSpan.FromSeconds(5)); + } + + [Fact] + public void A_psk_miss_costs_only_that_node_and_never_becomes_a_fan_level_abort() + { + // The dialer refuses a keyless cluster locally, so it arrives here as an ordinary per-node + // failure. The fan-level assertion is that the other cluster is still dialled. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", 4056, true, false)); + var dialer = new RecordingDialer + { + FailClusters = ["C2"], + Fault = new Grpc.Core.RpcException( + new Grpc.Core.Status(Grpc.Core.StatusCode.Unauthenticated, "no key")), + }; + var rig = SpawnGrpcRig(factory, dialer); + AwaitContactsLoaded(rig); + + EventFilter.Warning(contains: "Unauthenticated").ExpectOne(TimeSpan.FromSeconds(5), () => + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()))); + + AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); + } + + [Fact] + public void A_command_with_no_grpc_target_is_dropped_with_a_warning() + { + var factory = NewInMemoryDbFactory(); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () => + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()))); + + rig.Dialer.Calls.ShouldBeEmpty(); + } + + [Fact] + public void A_payload_that_is_not_a_mesh_command_record_is_dropped_with_an_error() + { + // The mapper refuses to encode an unknown payload rather than shipping something the node + // cannot decode. That refusal must cost the command, not the actor. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + EventFilter.Error(contains: "Cannot encode").ExpectOne(TimeSpan.FromSeconds(5), () => + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, "not a command record"))); + + rig.Dialer.Calls.ShouldBeEmpty(); + + // Still serving. + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); + } + + [Fact] + public void The_envelope_carries_the_commands_own_correlation_and_the_whole_fan_shares_it() + { + 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); + + var correlationId = new CorrelationId(Guid.NewGuid()); + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, new DispatchDeployment( + new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), correlationId))); + + AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); + rig.Dialer.Calls.ShouldAllBe(c => c.Envelope.CorrelationId == correlationId.ToString()); + // One envelope object, encoded once and fanned — not re-encoded per node. + rig.Dialer.Calls.Select(c => c.Envelope).Distinct().Count().ShouldBe(1); + } + + [Fact] + public void An_alarm_command_gets_one_minted_transport_correlation_shared_by_the_whole_fan() + { + // AlarmCommand is the one record with no correlation member of its own, so the envelope's is + // the only handle an operator has on it. Minting per fan rather than per node is what makes + // "this alarm command" one traceable event instead of N unrelated ones. + 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); + + rig.Actor.Tell(new MeshCommand( + "alarms", new AlarmCommand("alarm-1", "Acknowledge", "operator", null, null))); + + AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5)); + rig.Dialer.Calls.Select(c => c.Envelope.CorrelationId).Distinct().Count().ShouldBe(1); + rig.Dialer.Calls[0].Envelope.CorrelationId.ShouldNotBeNullOrWhiteSpace(); + rig.Dialer.Calls[0].Envelope.AlarmCommand.AlarmId.ShouldBe("alarm-1"); + } + + [Fact] + public void The_contact_refresh_publishes_the_known_cluster_set_and_drops_cached_psks() + { + // Phase 1A security register item 1, wired: the gate is only a control if something keeps it + // current, and the refresh is the moment central's picture of the fleet is up to date. The + // InvalidateAll on the same tick is what makes a rotated key take effect within one refresh + // instead of requiring a central restart. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", null, true, false)); + var rig = SpawnGrpcRig(factory); + + AwaitContactsLoaded(rig); + + // C2 is included even though its only node has no GrpcPort: the gate is about which clusters + // central KNOWS, not which it can currently dial. Excluding it would deny that cluster's acks. + rig.Psk.KnownClusterPublications[0].ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true); + rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1); + } + + [Fact] + public void A_slow_dial_does_not_park_the_actors_mailbox() + { + // The dialer's first act is a PSK lookup, which blocks on the secret store — bounded, but + // seconds, and repeated after every refresh because the refresh drops the cache. An async + // method runs synchronously to its first await, so dialing straight from the receive would + // stall this actor for that long PER NODE, and deployment acks queue behind that mailbox. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var coordinator = CreateTestProbe(); + var dialer = new RecordingDialer { BlockBeforeFirstAwait = true }; + var rig = SpawnGrpcRig(factory, dialer, coordinator: coordinator.Ref); + AwaitContactsLoaded(rig); + + try + { + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + // The ack must be handled while the dial is still parked. + var ack = new ApplyAck( + new DeploymentId(Guid.NewGuid()), + new NodeId("host-a:4053"), + ApplyAckOutcome.Applied, + null, + new CorrelationId(Guid.NewGuid())); + rig.Actor.Tell(ack); + + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).ShouldBe(ack); + } + finally + { + dialer.Release(); + } + } + + [Fact] + public void A_node_added_after_start_joins_the_fan_on_the_next_refresh() + { + // The refresh is not decoration: operators add and retire nodes, and central must pick them + // up without a restart. Under Grpc that is the ONLY thing keeping the dispatch set current — + // there is no receptionist to notice a new registration on its own. + // + // Driven by telling the actor its own RefreshContacts tick rather than by shortening + // ContactRefreshSeconds: a one-second timer would run for the whole fact, adding DB load to + // every test sharing the runner, and would make this assertion depend on wall-clock timing. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var rig = SpawnGrpcRig(factory); + AwaitContactsLoaded(rig); + + using (var db = factory.CreateDbContext()) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = "host-b:4053", + ClusterId = "C1", + Host = "host-b", + AkkaPort = 4053, + GrpcPort = 4056, + ApplicationUri = "urn:host-b", + Enabled = true, + CreatedBy = "test", + }); + db.SaveChanges(); + } + + rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts()); + AwaitCondition( + () => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5)); + + rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); + + 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); + } + + [Fact] + public void The_known_cluster_set_is_not_published_by_the_other_two_modes() + { + // 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. + var factory = NewInMemoryDbFactory(); + SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false)); + var psk = new RecordingPskProvider(); + + Sys.ActorOf(CentralCommunicationActor.Props( + factory, + ClusterClientMode(), + new RecordingClientFactory(() => CreateTestProbe()), + () => null, + dialer: null, + pskProvider: psk)); + + ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + psk.KnownClusterPublications.ShouldBeEmpty(); + psk.InvalidateAllCount.ShouldBe(0); + } + + [Fact] + public void Grpc_mode_with_no_dialer_wired_drops_the_command_with_an_error() + { + // Mis-wiring, not misconfiguration: Mode=Grpc validates fine but the DI graph did not supply + // a dialer. Silence here would be a fleet-wide command outage with no symptom until a deploy + // stalled at its apply deadline two minutes later. + var factory = NewInMemoryDbFactory(); + SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); + var actor = Sys.ActorOf(CentralCommunicationActor.Props( + factory, + GrpcMode(), + new RecordingClientFactory(() => CreateTestProbe()), + () => null, + dialer: null, + pskProvider: null)); + + // Filtered on the DISPATCH message's exact phrase: PreStart logs its own error about the same + // mis-wiring, and a looser filter races it into a second match. + EventFilter.Error(contains: "no command dialer is wired").ExpectOne( + TimeSpan.FromSeconds(5), + () => actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()))); + } + + private static DispatchDeployment NewDispatch() => new( + new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid())); + + /// A spawned gRPC-mode actor plus the two seams the tests observe it through. + private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk); + + private GrpcRig SpawnGrpcRig( + IDbContextFactory factory, + RecordingDialer? dialer = null, + IMeshClusterClientFactory? clientFactory = null, + IActorRef? coordinator = null) + { + dialer ??= new RecordingDialer(); + var psk = new RecordingPskProvider(); + var actor = Sys.ActorOf(CentralCommunicationActor.Props( + factory, + GrpcMode(), + clientFactory ?? new RecordingClientFactory(() => CreateTestProbe()), + () => coordinator, + dialer, + psk)); + return new GrpcRig(actor, dialer, psk); + } + + /// + /// Waits until the actor has applied a contact load. The known-cluster publication is the + /// signal because it happens in the same handler, immediately after the target set is + /// adopted — and unlike the target set itself it is observable from outside the actor. A + /// command told before this point would be dropped as "no targets" and never retried, which + /// is correct behaviour and a flaky test. + /// + private void AwaitContactsLoaded(GrpcRig rig) => + AwaitCondition( + () => rig.Psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5)); + + /// Records every dial and answers however the test asks; never opens a socket. + private sealed class RecordingDialer : IMeshCommandDialer + { + private readonly object _gate = new(); + private readonly List _calls = []; + private readonly ManualResetEventSlim _released = new(false); + + /// + /// When true, every dial blocks its calling thread synchronously until + /// . Models the dialer's pre-await PSK lookup. + /// + public bool BlockBeforeFirstAwait { get; set; } + + /// Unblocks every dial parked by . + public void Release() => _released.Set(); + + /// Every dial made. Copied under the lock — the fan runs on thread-pool threads. + public IReadOnlyList Calls + { + get { lock (_gate) return _calls.ToList(); } + } + + /// Endpoints whose substring match makes the dial fault. + public string? FailEndpointsContaining { get; set; } + + /// Cluster ids whose dials fault. + public string[] FailClusters { get; set; } = []; + + /// + /// The fault raised for a matching dial. Fully qualified because this file's bare + /// Status is Akka's — an existing fact drives the actor's Status.Failure + /// handler, and importing Grpc.Core here would make that reference ambiguous. + /// + public Grpc.Core.RpcException? Fault { get; set; } + + public Task DispatchAsync( + string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct) + { + lock (_gate) _calls.Add(new Dial(nodeEndpoint, envelope, clusterId)); + + if (BlockBeforeFirstAwait) + { + // Blocks the CALLING thread, exactly as IMeshPskProvider.TryGetPsk does on a cold + // cache: an async method runs synchronously to its first await, so a dial invoked + // straight from the actor would park the mailbox here. + _released.Wait(TimeSpan.FromSeconds(30)); + } + + var shouldFail = + (FailEndpointsContaining is not null && nodeEndpoint.Contains(FailEndpointsContaining)) + || FailClusters.Contains(clusterId, StringComparer.OrdinalIgnoreCase); + + return shouldFail && Fault is not null + ? Task.FromException(Fault) + : Task.CompletedTask; + } + + /// One recorded dial. + public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId); + } + + /// Records the gate and invalidation calls the contact refresh is supposed to make. + private sealed class RecordingPskProvider : IMeshPskProvider + { + private readonly object _gate = new(); + private readonly List> _publications = []; + private int _invalidateAllCount; + + /// Every known-cluster set published, in order. + public IReadOnlyList> KnownClusterPublications + { + get { lock (_gate) return _publications.ToList(); } + } + + /// How many times the whole cache was dropped. + public int InvalidateAllCount => Volatile.Read(ref _invalidateAllCount); + + public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk) + { + psk = "psk"; + return true; + } + + public bool Matches(string clusterId, string presented) => true; + + public void SetKnownClusters(IReadOnlyCollection clusterIds) + { + lock (_gate) _publications.Add(clusterIds.ToList()); + } + + public void Invalidate(string clusterId) { } + + public void InvalidateAll() => Interlocked.Increment(ref _invalidateAllCount); + } } 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 new file mode 100644 index 00000000..96a0db0f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs @@ -0,0 +1,271 @@ +using System.Diagnostics.CodeAnalysis; +using Grpc.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; + +/// +/// ClusterClient→gRPC Phase 1A — central's outbound command dial. One node, one call, no retry. +/// +/// +/// The facts that matter are the credential ones: the bearer must be the TARGET node's cluster's +/// key (not central's, and not some other cluster's), and a cluster with no resolvable key must +/// never be dialled at all. Everything else is plumbing. +/// +public sealed class MeshGrpcCommandDialerTests +{ + private const string Endpoint = "http://site-a-1:4056"; + + private static CommandEnvelope Envelope() => new() + { + CorrelationId = Guid.NewGuid().ToString("N"), + RestartDriver = new RestartDriverCommand + { + ClusterId = "SITE-A", + DriverInstanceId = "drv-1", + ActorByUserName = "operator", + CorrelationId = Guid.NewGuid().ToString("N"), + }, + }; + + private static MeshGrpcCommandDialer CreateDialer( + IMeshPskProvider psk, + RecordingClient client, + int dispatchTimeoutSeconds = 10) + => new( + psk, + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = dispatchTimeoutSeconds }), + _ => client); + + [Fact] + public async Task Dial_presents_the_target_clusters_key_and_names_that_cluster() + { + // THE assertion. Central holds N keys; sending the wrong one turns a legitimate command into + // a PermissionDenied, and sending a fleet-wide one would be the blast radius this design + // exists to avoid. + var client = new RecordingClient(); + var dialer = CreateDialer( + new FakePskProvider(("SITE-A", "psk-site-a"), ("SITE-B", "psk-site-b")), client); + + await dialer.DispatchAsync(Endpoint, Envelope(), "SITE-B", CancellationToken.None); + + client.LastHeaders.ShouldNotBeNull(); + client.LastHeaders!.GetValue("authorization").ShouldBe("Bearer psk-site-b"); + client.LastHeaders.GetValue("x-otopcua-cluster").ShouldBe("SITE-B"); + } + + [Fact] + public async Task A_cluster_with_no_key_is_refused_locally_and_never_dialled() + { + // Fail-closed: not "dial and let the node reject it". An unauthenticated command attempt + // against a node's control surface should not exist even as a rejected one. + var client = new RecordingClient(); + var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client); + + var ex = await Should.ThrowAsync( + () => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-Z", CancellationToken.None)); + + ex.StatusCode.ShouldBe(StatusCode.Unauthenticated); + client.Calls.ShouldBe(0); + } + + [Fact] + public async Task The_envelope_reaches_the_node_unchanged() + { + var client = new RecordingClient(); + var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client); + var envelope = Envelope(); + + await dialer.DispatchAsync(Endpoint, envelope, "SITE-A", CancellationToken.None); + + client.LastRequest.ShouldBe(envelope); + } + + [Fact] + public async Task A_node_rejection_surfaces_as_its_own_RpcException() + { + // The actor grades per-node outcomes off the status code, so it must arrive intact rather + // than wrapped in an AggregateException or swallowed here. + var client = new RecordingClient + { + Fault = new RpcException(new Status(StatusCode.PermissionDenied, "nope")), + }; + var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client); + + var ex = await Should.ThrowAsync( + () => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None)); + + ex.StatusCode.ShouldBe(StatusCode.PermissionDenied); + } + + [Fact] + public async Task The_per_call_deadline_bounds_a_node_that_never_answers() + { + // A dispatch dials every node in the fleet. Without this bound one unreachable node holds a + // task and a socket for as long as the process lives, while the deploy path's own apply + // deadline expires around it. + var client = new RecordingClient { HangUntilCancelled = true }; + var dialer = CreateDialer( + new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 1); + + await Should.ThrowAsync( + () => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None) + .WaitAsync(TimeSpan.FromSeconds(10))); + } + + [Fact] + public async Task The_callers_token_still_cancels_a_dial_inside_the_deadline() + { + // Linked, not replaced: actor shutdown must cancel everything in flight even when the + // per-call deadline is nowhere near expiring. + var client = new RecordingClient { HangUntilCancelled = true }; + var dialer = CreateDialer( + new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 600); + using var cts = new CancellationTokenSource(); + + var dial = dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", cts.Token); + await cts.CancelAsync(); + + await Should.ThrowAsync(() => dial.WaitAsync(TimeSpan.FromSeconds(10))); + } + + [Fact] + public async Task A_zero_dispatch_timeout_leaves_the_dial_bounded_only_by_the_callers_token() + { + // 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); + + await dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None); + + client.Calls.ShouldBe(1); + } + + [Fact] + public async Task Dialing_after_disposal_throws_rather_than_repopulating_the_channel_cache() + { + var client = new RecordingClient(); + var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client); + dialer.Dispose(); + + await Should.ThrowAsync( + () => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None)); + } + + [Fact] + public async Task The_real_channel_factory_dials_and_is_disposable() + { + // The fake client seam covers every other fact here, which means nothing else ever exercises + // GrpcChannel.ForAddress or the cache it fills. Dial an endpoint nothing is listening on: + // the failure must be an ordinary RpcException the actor can log, not a construction throw, + // 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 })); + + await Should.ThrowAsync( + () => dialer.DispatchAsync( + "http://127.0.0.1:1", Envelope(), "SITE-A", CancellationToken.None)); + + Should.NotThrow(dialer.Dispose); + } + + [Fact] + public void It_activates_from_the_container_despite_the_optional_client_factory_parameter() + { + // The seam is an optional constructor parameter, which MS.DI honours only because it checks + // HasDefaultValue. Get that wrong — by making it non-optional, or by adding a second public + // 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.AddSingleton(new FakePskProvider(("SITE-A", "psk-site-a"))); + services.AddSingleton>( + Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 10 })); + services.AddSingleton(); + + using var provider = services.BuildServiceProvider(); + + provider.GetRequiredService().ShouldBeOfType(); + } + + /// + /// A that records what it was + /// handed and answers however the test asks. The generated client's protected parameterless + /// constructor exists precisely for this. + /// + private sealed class RecordingClient : NodeCommandService.NodeCommandServiceClient + { + /// Number of calls that reached the client. + public int Calls { get; private set; } + + /// Headers of the most recent call. + public Metadata? LastHeaders { get; private set; } + + /// Request of the most recent call. + public CommandEnvelope? LastRequest { get; private set; } + + /// When set, the call faults with this instead of answering. + public RpcException? Fault { get; set; } + + /// When true, the call never completes until its token is cancelled. + public bool HangUntilCancelled { get; set; } + + public override AsyncUnaryCall ExecuteAsync( + CommandEnvelope request, + Metadata? headers = null, + DateTime? deadline = null, + CancellationToken cancellationToken = default) + { + Calls++; + LastHeaders = headers; + LastRequest = request; + + var response = Fault is not null + ? Task.FromException(Fault) + : HangUntilCancelled + ? HangAsync(cancellationToken) + : Task.FromResult(new ExecuteAck()); + + return new AsyncUnaryCall( + response, + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => [], + () => { }); + } + + private static async Task HangAsync(CancellationToken ct) + { + await Task.Delay(Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false); + return new ExecuteAck(); + } + } + + /// An over a fixed key set; unknown clusters fail closed. + private sealed class FakePskProvider(params (string Cluster, string Psk)[] keys) : IMeshPskProvider + { + private readonly Dictionary _keys = + keys.ToDictionary(k => k.Cluster, k => k.Psk, StringComparer.OrdinalIgnoreCase); + + public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk) + => _keys.TryGetValue(clusterId, out psk); + + public bool Matches(string clusterId, string presented) + => _keys.TryGetValue(clusterId, out var expected) + && string.Equals(expected, presented, StringComparison.Ordinal); + + public void SetKnownClusters(IReadOnlyCollection clusterIds) { } + + public void Invalidate(string clusterId) { } + + public void InvalidateAll() { } + } +}