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; + } + } }