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() { } + } +}