From 308c6aaa0f7cf1ff6202c2ba9138cc9dab703623 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 6 Aug 2026 05:48:29 -0400 Subject: [PATCH] feat(mesh): gRPC ApplyAck path behind Mode=Grpc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 1B.2 — the node→central half of the ClusterClient→gRPC migration. GrpcApplyAckSender walks MeshTransport:CentralGrpcEndpoints in order and STOPS AT THE FIRST SUCCESS. That single `return` is the whole single-count guarantee: the transport it replaces was ClusterClient.Send, deliberately not SendToAll, because both central nodes forward to the SAME ConfigPublishCoordinator singleton proxy — an ack posted to both would have the coordinator count one node twice and could seal a deployment on a doubled count. The list is an availability list, not a fan-out list. No retry beyond that one ordered pass, no queue: the receptionist buffered nothing and retried nothing, and this is like-for-like, not an upgrade to delivery guarantees. All endpoints failing is still a DROP — but a loud one, naming the node, deployment, correlation id and the status each endpoint answered with, which is what separates "central is down" from "central rejected this node's key". The bearer never reaches a log; only endpoints and status codes do. Per-endpoint GrpcChannel cache and the endpoint loop clone GrpcDeploymentArtifactFetcher; the cluster id in x-otopcua-cluster comes from this node's own cluster- role via RoleParser, off the same bound Cluster:Roles ClusterRoleInfo reads. SendAsync never throws: the caller discards the task, so an escaping exception would become an unobserved fault — a drop with no log line, the exact failure mode this migration removes. NodeCommunicationActor gains an OPTIONAL ackSender; every existing test compiles and passes unedited. It is checked before the ClusterClient so the two can never both fire. The registration arm creates no ClusterClient under Grpc. NOTE — one wiring change beyond the actor: DriverHostActor's ackRouter is now `useClusterClient || useGrpc ? nodeComm : null`. Left at ClusterClient-only, a Grpc node's ack would have gone out on the DPS deployment-acks topic and never reached the sender at all. --- .../Communication/GrpcApplyAckSender.cs | 203 ++++++++++ .../Communication/IApplyAckSender.cs | 36 ++ .../Communication/NodeCommunicationActor.cs | 35 +- .../ServiceCollectionExtensions.cs | 61 ++- .../Communication/GrpcApplyAckSenderTests.cs | 354 ++++++++++++++++++ .../NodeCommunicationActorTests.cs | 99 +++++ 6 files changed, 775 insertions(+), 13 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs new file mode 100644 index 00000000..6aa5b497 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs @@ -0,0 +1,203 @@ +using System.Collections.Concurrent; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; + +/// +/// Posts this node's to central's CentralCommandService as a unary +/// gRPC call (ClusterClient→gRPC migration Task 1B.2), the MeshTransport:Mode = Grpc +/// replacement for the ClusterClient Send. +/// +/// +/// +/// One ordered pass, stopping at the first success. +/// is an availability list, not a +/// fan-out list: either central node hands the ack to the same +/// ConfigPublishCoordinator singleton proxy, so posting to both would have the +/// coordinator count this node twice — and a doubled count can seal a deployment other nodes +/// never applied. That is exactly why the transport being replaced used +/// ClusterClient.Send and not SendToAll. +/// +/// +/// No retry beyond that single pass, and no queue. The receptionist buffered nothing +/// and retried nothing, and this is a like-for-like replacement, not an upgrade to delivery +/// guarantees. The deploy path's own apply deadline stays the only recovery mechanism; a +/// retry loop here would instead pile acks up behind an unreachable central and deliver a +/// burst of stale ones when it returned. +/// +/// +/// All endpoints failing is a DROP — but a loud one. That is the same outcome the +/// ClusterClient path had when it could not reach central; what changes is that the operator +/// now gets the node, the deployment, the correlation id and the status each endpoint +/// answered with, which is what distinguishes "central is down" from "central rejected this +/// node's key". +/// +/// +/// The bearer never reaches a log. It travels in the call headers on every attempt and +/// the failure log names endpoints and status codes only — a per-attempt dump of the +/// would write the cluster's pre-shared key into the node log of every +/// deployment that could not reach central. +/// +/// +/// The cluster id comes from this node's own cluster-<id> Akka role, read +/// through off the same bound Cluster:Roles that +/// ClusterRoleInfo reads (the OTOPCUA_ROLES fallback is already applied to those +/// options by AddOtOpcUaCluster's PostConfigure, so nothing is re-resolved +/// here). The two must agree: central grades the bearer against the key of the cluster named +/// in the header, so a node that authenticated as a different cluster than it gossips would be +/// rejected on every ack. +/// +/// +public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable +{ + /// The cluster the caller claims to be; central grades the bearer against ITS key. + private const string ClusterHeader = "x-otopcua-cluster"; + + private readonly MeshTransportOptions _options; + private readonly string? _clusterId; + private readonly ILogger _log; + private readonly Func _clientFactory; + private readonly ConcurrentDictionary _channels = new(StringComparer.Ordinal); + + /// Initializes a new instance of the class. + /// Mesh-transport options; supplies the endpoints, the PSK and the deadline. + /// + /// Cluster options, read only for this node's roles — the cluster-<id> one names + /// the cluster this node authenticates as. + /// + /// Logger. + /// + /// Seam for tests: builds the generated client for an endpoint. Defaults to a real + /// (one cached per endpoint). + /// + public GrpcApplyAckSender( + IOptions options, + IOptions clusterOptions, + ILogger log, + Func? clientFactory = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(clusterOptions); + ArgumentNullException.ThrowIfNull(log); + + _options = options.Value; + _log = log; + _clusterId = RoleParser.ClusterIdFromRoles(clusterOptions.Value.Roles); + _clientFactory = clientFactory ?? DefaultClientFactory; + + if (_clusterId is null) + { + // Said once, at construction, rather than on every ack. Central cannot grade a bearer it + // has no cluster for, so every ack this node posts will come back PermissionDenied — and + // without this line that reads as a key mismatch rather than a missing role. + _log.LogWarning( + "This node carries no '{Prefix}' role, so its deployment acks will be " + + "posted without the {Header} header and central will reject every one of them. Add " + + "the cluster-scoped role to Cluster:Roles (or OTOPCUA_ROLES).", + RoleParser.ClusterRolePrefix, ClusterHeader); + } + } + + /// + public async Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(ack); + + try + { + await PostAsync(ack, cancellationToken); + } + catch (Exception ex) + { + // Deliberately unbounded, and deliberately not a rethrow. The caller discards this task + // (see IApplyAckSender), so anything escaping would be swallowed by the runtime and + // resurface as an unobserved exception at GC — a drop with no line in the node log, which + // is precisely the failure mode this transport replaces. Encoding faults and non-RpcException + // transport faults both land here. + _log.LogWarning( + ex, + "Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: " + + "posting it to central failed unexpectedly. It is not buffered and will not be " + + "retried, so the deployment will time out at central naming this node even though it " + + "applied successfully.", + ack.DeploymentId, ack.CorrelationId, ack.NodeId); + } + } + + private async Task PostAsync(ApplyAck ack, CancellationToken cancellationToken) + { + var endpoints = _options.CentralGrpcEndpoints; + if (endpoints.Length == 0) + { + _log.LogWarning( + "Dropping the ApplyAck for {DeploymentId} from {NodeId}: no " + + "MeshTransport:CentralGrpcEndpoints are configured, so there is nowhere to post it. " + + "The deployment will time out at central naming this node even though it applied.", + ack.DeploymentId, ack.NodeId); + return; + } + + var message = MeshTransportDtoMapper.ToProto(ack); + var headers = new Metadata { { "authorization", $"Bearer {_options.GrpcPsk}" } }; + if (_clusterId is not null) + headers.Add(ClusterHeader, _clusterId); + + var deadline = _options.AckTimeoutSeconds > 0 + ? DateTime.UtcNow.AddSeconds(_options.AckTimeoutSeconds) + : (DateTime?)null; + + var failures = new List(endpoints.Length); + foreach (var endpoint in endpoints) + { + try + { + var client = _clientFactory(endpoint); + await client.ReportApplyAsync(message, headers, deadline, cancellationToken); + + _log.LogDebug( + "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint}.", + ack.DeploymentId, ack.Outcome, endpoint); + + // The single-count guarantee, and it is this `return`: the pass stops at the first + // endpoint that accepted the ack and never walks back. + return; + } + catch (RpcException ex) + { + // Status code only — never the exception's own message and never the headers, either + // of which could echo credential material back into the log. + failures.Add($"{endpoint} => {ex.StatusCode}"); + } + } + + _log.LogWarning( + "Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: all " + + "{Count} central endpoint(s) refused it ({Failures}). It is not buffered and will not be " + + "retried, so the deployment will time out at central naming this node even though it " + + "applied successfully.", + ack.DeploymentId, ack.CorrelationId, ack.NodeId, endpoints.Length, string.Join("; ", failures)); + } + + private CentralCommandService.CentralCommandServiceClient DefaultClientFactory(string endpoint) + { + // h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. One channel per + // endpoint, cached and reused (creating a channel per ack would leak a socket per deployment). + var channel = _channels.GetOrAdd(endpoint, ep => GrpcChannel.ForAddress(ep)); + return new CentralCommandService.CentralCommandServiceClient(channel); + } + + /// + public void Dispose() + { + foreach (var channel in _channels.Values) + channel.Dispose(); + _channels.Clear(); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs new file mode 100644 index 00000000..ca631ea6 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs @@ -0,0 +1,36 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; + +/// +/// Posts this node's deployment acknowledgment to central over the mesh_transport.v1 +/// command plane (ClusterClient→gRPC migration Phase 1B). The seam +/// uses under MeshTransport:Mode = Grpc, in place of +/// the ClusterClient.Send it uses under ClusterClient. +/// +/// +/// +/// One ack per node per dispatch. Whatever the implementation does about redundancy, it +/// must deliver the ack to at most one central endpoint: both central nodes forward to +/// the same ConfigPublishCoordinator singleton proxy, so a fan-out would have the +/// coordinator count one node twice. That is the semantic of the ClusterClient.Send +/// this replaces — Send, deliberately not SendToAll. +/// +/// +/// Fire-and-forget, and it never throws. The caller is an actor that discards the +/// returned task, so an exception escaping here would become an unobserved task fault — the +/// silent drop this migration exists to remove. An implementation reports failure by logging +/// and returning, never by faulting the task. +/// +/// +public interface IApplyAckSender +{ + /// Posts to central, or logs why it could not be. + /// The per-node apply acknowledgment. + /// Cancels the in-flight call. + /// + /// A task that completes when the attempt finishes, successfully or not. It never faults — + /// see the remarks. + /// + Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs index fe192b2d..dfa46a12 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs @@ -37,23 +37,31 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; public sealed class NodeCommunicationActor : ReceiveActor { private readonly IActorRef? _centralClient; + private readonly IApplyAckSender? _ackSender; private readonly ILoggingAdapter _log = Context.GetLogger(); /// Creates the props for this actor. /// /// ClusterClient dialling central's receptionist, or under /// MeshTransport:Mode = Dps, where acks still go out over DistributedPubSub from - /// DriverHostActor and this actor never sees them. + /// DriverHostActor and this actor never sees them — and under + /// Mode = Grpc, where no ClusterClient is created at all. + /// + /// + /// The gRPC ack transport under MeshTransport:Mode = Grpc, or + /// under Dps and ClusterClient. /// /// The props. - public static Props Props(IActorRef? centralClient) => - Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient)); + public static Props Props(IActorRef? centralClient, IApplyAckSender? ackSender = null) => + Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient, ackSender)); /// Initializes a new instance of the class. /// ClusterClient dialling central, or . - public NodeCommunicationActor(IActorRef? centralClient) + /// The gRPC ack transport, or . + public NodeCommunicationActor(IActorRef? centralClient, IApplyAckSender? ackSender = null) { _centralClient = centralClient; + _ackSender = ackSender; // Inbound from central. Each type is listed explicitly rather than caught by a ReceiveAny: // an unknown type must dead-letter loudly during development rather than be republished @@ -75,13 +83,28 @@ public sealed class NodeCommunicationActor : ReceiveActor private void HandleApplyAck(ApplyAck ack) { + if (_ackSender is not null) + { + // Fire-and-forget by construction, exactly as the ClusterClient Tell below is. The sender + // never throws and never retries (see IApplyAckSender), so there is nothing to await and + // nothing to pipe back: awaiting would stall this mailbox behind a network call to central, + // and an Ask would invent a request/response the one-way contract does not have. + // + // Checked BEFORE the ClusterClient. The registration arm builds exactly one transport, so + // both are never set — but were they, posting over both would deliver two acks for one + // apply and the coordinator would count this node twice. + _ = _ackSender.SendAsync(ack); + return; + } + if (_centralClient is null) { _log.Warning( - "No central ClusterClient — dropping ApplyAck for {DeploymentId}. The ack is not " + "No central ack transport — dropping ApplyAck for {DeploymentId}. The ack is not " + "buffered and will not be retried, so the deployment will time out at central " + "naming this node even though it applied successfully. Check " - + "MeshTransport:CentralContactPoints", + + "MeshTransport:CentralContactPoints (ClusterClient mode) or " + + "MeshTransport:CentralGrpcEndpoints (Grpc mode)", ack.DeploymentId); return; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index eeda4156..dd7c05ff 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using Akka.Actor; using Akka.Cluster.Tools.Client; +using Akka.Event; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; @@ -99,6 +100,26 @@ public static class ServiceCollectionExtensions // enabled, so piggybacking would leave a redundant pair that historizes values but not alarms // with no gate at all, and both nodes would write every sample twice with nothing to say so. services.TryAddSingleton(sp => sp.GetRequiredService()); + + // ClusterClient→gRPC migration Task 1B.2: the node→central ack transport under + // MeshTransport:Mode = Grpc. Registered unconditionally but resolved only by the Grpc arm of + // WithOtOpcUaRuntimeActors — the mode is not knowable here (this runs before configuration is + // read into options), and a singleton nobody resolves is never constructed, so a Dps node + // opens no channel. + // + // Registered in the CONTAINER rather than new-ed up beside the actor so its per-endpoint + // GrpcChannel cache is disposed at host shutdown; an actor-scoped instance would leak the + // channels on every restart. + // + // Built by a factory, not by type: MS.DI does not honour optional constructor parameters, so + // the clientFactory test seam would make type-based activation fail outright. The logger is + // resolved leniently for the same reason DriverHostActor's is — a harness that skipped logging + // should not fail to build the graph. + services.TryAddSingleton(sp => new GrpcApplyAckSender( + sp.GetRequiredService>(), + sp.GetRequiredService>(), + sp.GetService()?.CreateLogger() + ?? NullLogger.Instance)); return services; } @@ -389,6 +410,8 @@ public static class ServiceCollectionExtensions var meshOptions = resolver.GetService>().Value; var useClusterClient = string.Equals( meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); + var useGrpc = string.Equals( + meshOptions.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); IActorRef? centralClient = null; if (useClusterClient) @@ -406,12 +429,33 @@ public static class ServiceCollectionExtensions CentralClusterClientName); } + // ClusterClient→gRPC migration Task 1B.2. Deliberately EXCLUSIVE with the branch above — + // under Grpc no ClusterClient is created at all, which is half the point of the migration + // (it is the cross-WAN Akka association this program removes). The two are never both + // handed to the actor, and it prefers the sender if they ever were. + IApplyAckSender? ackSender = null; + if (useGrpc) + { + ackSender = resolver.GetService(); + if (ackSender is null) + { + // Reachable only in a graph that skipped AddOtOpcUaRuntime (a test harness), never + // in the Host. Said here rather than left to the actor's per-ack drop warning, + // because the cause is a missing registration and the symptom would otherwise + // appear once per deployment with no hint of why. + system.Log.Warning( + "MeshTransport:Mode is {Mode} but no {Sender} is registered — every ApplyAck " + + "this node produces will be dropped. AddOtOpcUaRuntime registers it.", + meshOptions.Mode, nameof(IApplyAckSender)); + } + } + var nodeComm = system.ActorOf( - NodeCommunicationActor.Props(centralClient), + NodeCommunicationActor.Props(centralClient, ackSender), MeshPaths.NodeCommunicationName); - // Registered with the receptionist in BOTH modes: under Dps nothing dials it, and an - // idle registration costs nothing — whereas registering only under ClusterClient would + // Registered with the receptionist in ALL modes: under Dps and Grpc nothing dials it, and + // an idle registration costs nothing — whereas registering only under ClusterClient would // make flipping the flag require a node restart before central could reach it. ClusterClientReceptionist.Get(system).RegisterService(nodeComm); registry.Register(nodeComm); @@ -523,9 +567,12 @@ public static class ServiceCollectionExtensions : null; var driverHost = system.ActorOf( - // ackRouter: under ClusterClient mode the node comm actor relays ApplyAcks across - // the boundary; under Dps it stays null and the host publishes on the - // deployment-acks topic exactly as before. + // ackRouter: under ClusterClient AND Grpc the node comm actor relays ApplyAcks across + // the boundary (which transport it then uses is its own business — the ClusterClient + // ref or the gRPC sender it was built with); under Dps it stays null and the host + // publishes on the deployment-acks topic exactly as before. Grpc MUST be included + // here: routed to DPS instead, a Grpc node's ack would never reach the sender at all + // and every deployment to it would time out while it applied perfectly. // dbFactory is nullable (Phase 4): a driver-only node passes null here — ConfigDb is // gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor // guards every ConfigDb-touching path on it (UpsertNodeDeploymentState no-ops; central @@ -533,7 +580,7 @@ public static class ServiceCollectionExtensions // alarmStateStore (there is no ConfigDb-backed EF fallback; the host skips the alarm // host when no store was wired). DriverHostActor.Props(dbFactory, roleInfo.LocalNode, - ackRouter: useClusterClient ? nodeComm : null, + ackRouter: useClusterClient || useGrpc ? nodeComm : null, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, dependencyMux: mux, opcUaPublishActor: publishActor, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs new file mode 100644 index 00000000..58befc3e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs @@ -0,0 +1,354 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication; + +/// +/// The node-side gRPC ack sender (ClusterClient→gRPC migration Task 1B.2). The client is a fake, +/// so what is under test is the sender's own logic: the ordered single-pass over +/// CentralGrpcEndpoints, the per-call credentials, and the drop that used to be silent. +/// +/// +/// +/// The load-bearing fact is that ONE endpoint is dialled on success. This replaces an +/// Akka ClusterClient.Send, which is single-target at-most-once — not a +/// SendToAll. Both central nodes route an ack to the same +/// ConfigPublishCoordinator singleton proxy, so posting to both would have the +/// coordinator count this node twice and could seal a deployment on a doubled quorum. +/// +/// +/// The failure side matters just as much: the receptionist dropped an undeliverable ack in +/// silence, and the whole point of the migration is that the same drop is now a Warning naming +/// every endpoint and the status it answered with. +/// +/// +public sealed class GrpcApplyAckSenderTests +{ + private const string Psk = "psk-that-must-never-be-logged"; + + private static ApplyAck Ack() => new( + new DeploymentId(Guid.NewGuid()), + new NodeId("site-a-1:4053"), + ApplyAckOutcome.Applied, + null, + new CorrelationId(Guid.NewGuid())); + + private static GrpcApplyAckSender Sender( + string[] endpoints, + Func factory, + ILogger? log = null, + string[]? roles = null) + => new( + Options.Create(new MeshTransportOptions + { + Mode = MeshTransportOptions.ModeGrpc, + CentralGrpcEndpoints = endpoints, + GrpcPsk = Psk, + AckTimeoutSeconds = 7, + }), + Options.Create(new AkkaClusterOptions { Roles = roles ?? ["driver", "cluster-SITE-A"] }), + log ?? NullLogger.Instance, + factory); + + [Fact] + public async Task A_successful_first_endpoint_is_the_ONLY_one_dialled() + { + // The single-count semantic. Two endpoints are configured for availability, not for fan-out: + // an ack posted to both would reach the one coordinator singleton twice. + var consulted = new List(); + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var factory = (string endpoint) => + { + consulted.Add(endpoint); + return (CentralCommandService.CentralCommandServiceClient)new RecordingClient(calls); + }; + + await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack()); + + consulted.ShouldBe(["http://central-1:4055"]); + calls.Count.ShouldBe(1); + } + + [Fact] + public async Task The_wire_message_carries_the_whole_ack() + { + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var ack = new ApplyAck( + new DeploymentId(Guid.NewGuid()), + new NodeId("site-a-2:4053"), + ApplyAckOutcome.Failed, + "disk full", + new CorrelationId(Guid.NewGuid())); + + await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(ack); + + var sent = calls.Single().Request; + sent.DeploymentId.ShouldBe(ack.DeploymentId.ToString()); + sent.NodeId.ShouldBe("site-a-2:4053"); + sent.Outcome.ShouldBe(ApplyOutcome.Failed); + sent.FailureReason.ShouldBe("disk full"); + sent.CorrelationId.ShouldBe(ack.CorrelationId.ToString()); + } + + [Fact] + public async Task Every_call_carries_the_bearer_the_cluster_header_and_a_deadline() + { + // Central's CentralCommandAuthInterceptor grades the bearer against the key of the cluster + // named in x-otopcua-cluster, so an ack missing either header is PermissionDenied — and the + // deadline is what stops a dead first endpoint from stalling the ack forever. + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var before = DateTime.UtcNow; + + await Sender(["http://central-1:4055"], _ => new RecordingClient(calls)).SendAsync(Ack()); + + var options = calls.Single().Options; + options.Headers.ShouldNotBeNull(); + options.Headers.GetValue("authorization").ShouldBe($"Bearer {Psk}"); + options.Headers.GetValue("x-otopcua-cluster").ShouldBe("SITE-A"); + options.Deadline.ShouldNotBeNull(); + options.Deadline!.Value.ShouldBeGreaterThan(before); + options.Deadline!.Value.ShouldBeLessThanOrEqualTo(before.AddSeconds(7).AddSeconds(1)); + } + + [Fact] + public async Task An_unavailable_first_endpoint_fails_over_to_the_second() + { + var consulted = new List(); + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var factory = (string endpoint) => + { + consulted.Add(endpoint); + return endpoint.Contains("central-1", StringComparison.Ordinal) + ? (CentralCommandService.CentralCommandServiceClient) + new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down"))) + : new RecordingClient(calls); + }; + + await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack()); + + consulted.ShouldBe(["http://central-1:4055", "http://central-2:4055"]); + calls.Count.ShouldBe(1); + } + + [Fact] + public async Task When_every_endpoint_fails_the_ack_is_dropped_with_one_naming_Warning() + { + // The pre-migration behaviour was a silent drop. It is still a drop — there is no queue and no + // retry — but the operator now gets the node, the deployment, the correlation id and what each + // endpoint actually answered, which is everything needed to tell "central is down" from + // "central rejected this node's key". + var log = new CapturingLogger(); + var attempts = 0; + var factory = (string endpoint) => + { + attempts++; + return (CentralCommandService.CentralCommandServiceClient)new ThrowingClient( + new RpcException(new Status( + endpoint.Contains("central-1", StringComparison.Ordinal) + ? StatusCode.Unavailable + : StatusCode.PermissionDenied, + "no"))); + }; + var ack = Ack(); + + await Sender(["http://central-1:4055", "http://central-2:4055"], factory, log).SendAsync(ack); + + attempts.ShouldBe(2); + var warning = log.Entries.Single(e => e.Level == LogLevel.Warning); + warning.Message.ShouldContain(ack.NodeId.Value!); + warning.Message.ShouldContain(ack.DeploymentId.ToString()); + warning.Message.ShouldContain(ack.CorrelationId.ToString()); + warning.Message.ShouldContain("Unavailable"); + warning.Message.ShouldContain("PermissionDenied"); + } + + [Fact] + public async Task A_failed_pass_never_posts_the_ack_twice() + { + // The counterpart of the first fact: failover walks forward only. A retry of an + // already-delivered endpoint could double-count this node at the coordinator, which is the one + // outcome worse than the ack being lost. + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var consulted = new List(); + var factory = (string endpoint) => + { + consulted.Add(endpoint); + return endpoint.Contains("c1", StringComparison.Ordinal) + ? (CentralCommandService.CentralCommandServiceClient) + new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down"))) + : new RecordingClient(calls); + }; + + await Sender(["http://c1:4055", "http://c2:4055", "http://c3:4055"], factory).SendAsync(Ack()); + + // c3 is never dialled: the pass stops at the first success, and it does not go back to c1. + consulted.ShouldBe(["http://c1:4055", "http://c2:4055"]); + calls.Count.ShouldBe(1); + } + + [Fact] + public async Task The_pre_shared_key_is_never_written_to_a_log() + { + // The bearer travels in the call headers on every attempt; the failure log names endpoints and + // status codes only. A per-endpoint dump of the Metadata would put the cluster's PSK in the + // node log of every deployment that could not reach central. + var log = new CapturingLogger(); + var factory = (string _) => (CentralCommandService.CentralCommandServiceClient) + new ThrowingClient(new RpcException(new Status(StatusCode.Unavailable, "down"))); + + await Sender(["http://central-1:4055"], factory, log).SendAsync(Ack()); + + log.Entries.ShouldNotBeEmpty(); + log.Entries.Any(e => e.Message.Contains(Psk, StringComparison.Ordinal)).ShouldBeFalse(); + } + + [Fact] + public async Task An_empty_endpoint_list_is_a_logged_drop_not_a_throw() + { + // The validator refuses to start a Grpc driver node with no endpoints, so this is reachable + // only through a code path that built the sender for a node the validator never graded (a test + // harness, an admin-only graph). It must still be a visible drop rather than an exception on a + // task the actor discards, which would vanish entirely. + var log = new CapturingLogger(); + var dialled = false; + var factory = (string _) => + { + dialled = true; + return (CentralCommandService.CentralCommandServiceClient)new RecordingClient([]); + }; + + await Sender([], factory, log).SendAsync(Ack()); + + dialled.ShouldBeFalse(); + log.Entries.Single(e => e.Level == LogLevel.Warning).Message + .ShouldContain("CentralGrpcEndpoints"); + } + + [Fact] + public async Task A_node_with_no_cluster_role_still_posts_and_is_warned_about() + { + // No cluster-scoped role means no x-otopcua-cluster header, and central has no key to grade the + // bearer against — every ack will come back PermissionDenied. Posting anyway (rather than + // refusing locally) keeps the rejection where it can be read on both sides, and the local + // Warning explains a PermissionDenied that would otherwise look like a key mismatch. + var log = new CapturingLogger(); + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + + await Sender(["http://central-1:4055"], _ => new RecordingClient(calls), log, roles: ["driver"]) + .SendAsync(Ack()); + + calls.Single().Options.Headers!.Any(h => + h.Key.Equals("x-otopcua-cluster", StringComparison.Ordinal)).ShouldBeFalse(); + log.Entries.Any(e => e.Level == LogLevel.Warning + && e.Message.Contains("cluster-", StringComparison.Ordinal)).ShouldBeTrue(); + } + + [Fact] + public async Task An_unexpected_client_fault_is_a_logged_drop_not_an_unobserved_task_fault() + { + // The caller is an actor that discards the returned Task, so anything escaping here would be + // swallowed by the runtime and rediscovered as an unobserved exception at GC — the exact + // silent drop this transport exists to remove. + var log = new CapturingLogger(); + var factory = (string _) => (CentralCommandService.CentralCommandServiceClient) + new FaultingClient(new InvalidOperationException("channel disposed")); + + await Should.NotThrowAsync(() => Sender(["http://central-1:4055"], factory, log).SendAsync(Ack())); + + log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue(); + } + + [Fact] + public void Dispose_is_idempotent() + { + var sender = Sender(["http://central-1:4055"], _ => new RecordingClient([])); + + sender.Dispose(); + Should.NotThrow(sender.Dispose); + } + + // ---- fakes -------------------------------------------------------------------------------- + + private static AsyncUnaryCall Ok() => new( + Task.FromResult(new ReportAck()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => [], + () => { }); + + /// A generated client double recording the request and the call options it was given. + private sealed class RecordingClient(List<(ApplyAckMessage Request, CallOptions Options)> calls) + : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + { + calls.Add((request, options)); + return Ok(); + } + } + + /// A generated client double whose call throws — an endpoint that is down or refuses. + private sealed class ThrowingClient(RpcException toThrow) : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + => throw toThrow; + } + + /// A generated client double throwing something that is NOT an . + private sealed class FaultingClient(Exception toThrow) : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + => throw toThrow; + } + + /// Test logger that captures all log entries. + /// The category type. + private sealed class CapturingLogger : ILogger + { + /// Gets the captured log entries. + public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = []; + + /// Begins a scope (stub implementation). + /// The type of the scope state. + /// The scope state. + /// A no-op scope. + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + + /// Checks if logging is enabled (always true). + /// The log level to check. + /// true. + public bool IsEnabled(LogLevel logLevel) => true; + + /// Logs an entry and captures it. + /// The type of the log state. + /// The log level. + /// The event ID. + /// The log state. + /// The exception, if any. + /// The log message formatter. + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + Entries.Add((logLevel, formatter(state, exception), exception)); + } + + private sealed class NullScope : IDisposable + { + /// Gets the singleton instance. + public static NullScope Instance { get; } = new(); + + /// No-op. + public void Dispose() { } + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs index 7cc06aca..fbbe0003 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs @@ -127,4 +127,103 @@ public class NodeCommunicationActorTests : RuntimeActorTestBase EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne( TimeSpan.FromSeconds(5), () => actor.Tell(ack)); } + + // ---- Mode=Grpc (ClusterClient→gRPC migration Task 1B.2) ------------------------------------- + + private static ApplyAck NewAck() => new( + new DeploymentId(Guid.NewGuid()), + new NodeId("host-a:4053"), + ApplyAckOutcome.Applied, + null, + new CorrelationId(Guid.NewGuid())); + + [Fact] + public void Under_Grpc_the_ack_goes_to_the_sender_exactly_once() + { + // The gRPC counterpart of the ClusterClient.Send fact above, and it carries the same + // single-count obligation: one post per node per dispatch, because both central endpoints + // route to the SAME coordinator singleton proxy. Where the ack goes among the endpoints is the + // sender's business (GrpcApplyAckSenderTests); what this pins is that the actor hands it over + // once and does not also fan it out itself. + var sender = new RecordingAckSender(); + var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender)); + + var ack = NewAck(); + actor.Tell(ack); + + AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5)); + sender.Sent[0].ShouldBe(ack); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + sender.Sent.Count.ShouldBe(1); + } + + [Fact] + public void A_sender_wins_over_a_cluster_client_and_the_ack_is_not_sent_twice() + { + // Defence in depth. The registration arm builds exactly one transport — under Grpc no + // ClusterClient is created at all — but if both were ever present, posting over both would + // deliver two acks for one apply and the coordinator would count this node twice. + var client = CreateTestProbe(); + var sender = new RecordingAckSender(); + var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref, sender)); + + actor.Tell(NewAck()); + + AwaitAssert(() => sender.Sent.Count.ShouldBe(1), TimeSpan.FromSeconds(5)); + client.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + [Fact] + public void The_ClusterClient_branch_never_touches_the_sender() + { + // Null-safe by construction: under Dps and ClusterClient the sender is not built at all, so the + // ApplyAck path must not depend on one existing. + var client = CreateTestProbe(); + var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref)); + + var ack = NewAck(); + actor.Tell(ack); + + client.ExpectMsg(TimeSpan.FromSeconds(5)).Message.ShouldBe(ack); + } + + [Fact] + public void Inbound_commands_are_unaffected_by_the_presence_of_a_sender() + { + // The sender is outbound-only. A Grpc node still re-emits central's commands on its own + // EventStream exactly as every other mode does. + var subscriber = CreateTestProbe(); + Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment)); + var sender = new RecordingAckSender(); + var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null, ackSender: sender)); + + var dispatch = NewDispatch(); + actor.Tell(dispatch); + + subscriber.ExpectMsg(TimeSpan.FromSeconds(5)).ShouldBe(dispatch); + sender.Sent.ShouldBeEmpty(); + } + + /// An double recording what the actor handed it. + private sealed class RecordingAckSender : IApplyAckSender + { + private readonly Lock _gate = new(); + private readonly List _sent = []; + + /// Gets a snapshot of the acks handed to this sender. + public IReadOnlyList Sent + { + get { lock (_gate) return [.. _sent]; } + } + + /// Records . + /// The ack the actor handed over. + /// Unused. + /// A completed task — the real sender never throws either. + public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) + { + lock (_gate) _sent.Add(ack); + return Task.CompletedTask; + } + } }