From 16d598560a3c86f5ec91c21b86352fd06a87d94e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 22 Jul 2026 16:42:07 -0400 Subject: [PATCH] feat(mesh): register comm actors with the receptionist; pin the wire-contract paths Phase 2 Task 6. Both comm actors are now spawned and receptionist-registered: CentralCommunicationActor from WithOtOpcUaControlPlaneSingletons, and NodeCommunicationActor from WithOtOpcUaRuntimeActors (before DriverHostActor, because it is the host's ackRouter under ClusterClient mode; it has no dependency of its own on the host, since inbound commands travel via the node's EventStream). Both are plain WithActors registrations, NOT singletons -- a driver node's ClusterClient rotates across contact points and must find a live comm actor at whichever node answers. Both register with the receptionist in BOTH transport modes: an idle registration costs nothing, whereas registering only under ClusterClient would mean flipping the flag on a running fleet needs a restart before anything can be reached, turning a config change back into a deployment. The coordinator ref is resolved lazily per message (Func + registry.TryGet) so the comm actor never depends on the order in which Akka.Hosting materialises the singleton proxy relative to this block. Adds MeshCommActorPathTests, which boots the real two-node host and Identify- probes both paths. These strings are the wire contract and nothing else asserts they agree: a rename would compile, pass every unit test, deploy, and deliver nothing, because a ClusterClient send to an unregistered path is dropped silently. Sabotage-verified with an actual rename. DELIBERATELY NOT TESTED, with the reason recorded in the file: that the actors are per-node rather than singletons. Two discriminators were tried and both were invalid -- "the path resolves on both nodes" passes under either shape because a ClusterSingletonManager is also created on every node in the role, and "no /singleton child" is null even for the KNOWN singleton /user/config-publish, so Akka.Hosting does not lay them out that way. A sabotage re-registering the actor via WithSingleton left both green, which is what exposed them. Rather than ship an assertion that cannot fail, the property is left to the Task 8 boundary test, where a send only arrives if the target really is registered. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../ServiceCollectionExtensions.cs | 38 ++++++++++++ .../ServiceCollectionExtensions.cs | 48 +++++++++++++- .../MeshCommActorPathTests.cs | 62 +++++++++++++++++++ 3 files changed, 147 insertions(+), 1 deletion(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 5fba90b4..73c7e6d7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -1,12 +1,17 @@ using Akka.Actor; using Akka.Cluster.Hosting; +using Akka.Cluster.Tools.Client; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations; using ZB.MOM.WW.OtOpcUa.ControlPlane.Audit; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet; using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy; @@ -112,6 +117,38 @@ public static class ServiceCollectionExtensions singletonOptions, createProxyToo: false); + // Per-cluster mesh Phase 2: the central end of the central↔node command boundary. + // + // A plain WithActors, NOT a singleton — and that is the load-bearing part. A driver node's + // ClusterClient rotates across contact points and must find a live comm actor at whichever + // central node answers. As a singleton it would exist on one central node only, and the + // rotation would fail silently against the other: acks would land sometimes, depending on + // which contact the client happened to settle on. + builder.WithActors((system, registry, resolver) => + { + var dbFactory = resolver.GetService>(); + var options = resolver.GetService>().Value; + + var actor = system.ActorOf( + CentralCommunicationActor.Props( + dbFactory, + options, + new DefaultMeshClusterClientFactory(), + // Resolved lazily, per message, so this actor never depends on the order in + // which Akka.Hosting materialises the singleton proxy relative to this block. + () => registry.TryGet(out var coordinator) + ? coordinator + : null), + MeshPaths.CentralCommunicationName); + + // Registered unconditionally, in both transport modes. Under Dps nothing dials it, and + // an idle registration costs nothing — whereas registering only under ClusterClient + // would mean flipping the flag on a running fleet requires a central restart before any + // node can ack, turning a config change back into a deployment. + ClusterClientReceptionist.Get(system).RegisterService(actor); + registry.Register(actor); + }); + return builder; } } @@ -123,3 +160,4 @@ public sealed class AuditWriterActorKey { } public sealed class FleetStatusBroadcasterKey { } public sealed class RedundancyStateActorKey { } public sealed class ClusterNodeAddressReconcilerKey { } +public sealed class CentralCommunicationActorKey { } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 0cb262c7..771e9ae0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -1,4 +1,6 @@ +using System.Collections.Immutable; using Akka.Actor; +using Akka.Cluster.Tools.Client; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; @@ -6,6 +8,10 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; using ZB.MOM.WW.OtOpcUa.Commons.Engines; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; @@ -37,6 +43,7 @@ public static class ServiceCollectionExtensions public const string OpcUaPublishActorName = "opcua-publish"; public const string PeerProbeSupervisorName = "peer-probe-supervisor"; public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder"; + public const string CentralClusterClientName = "central-cluster-client"; /// /// Registers shared runtime services. Currently binds @@ -319,6 +326,40 @@ public static class ServiceCollectionExtensions var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName); registry.Register(mux); + // Per-cluster mesh Phase 2: the node end of the central↔node command boundary. Spawned + // BEFORE DriverHostActor because it is the host's ackRouter under ClusterClient mode. + // It has no dependency of its own on the host — inbound commands travel via the node's + // EventStream — so this ordering costs nothing. + var meshOptions = resolver.GetService>().Value; + var useClusterClient = string.Equals( + meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); + + IActorRef? centralClient = null; + if (useClusterClient) + { + // Contacts come from appsettings, not the database — the deliberate asymmetry: + // central's node set changes as operators add and retire nodes, but central's own + // address is part of the deployment. MeshTransportOptionsValidator has already + // rejected an empty or malformed list at boot, so this cannot silently produce a + // client with no contacts. + var contacts = meshOptions.CentralContactPoints + .Select(cp => ActorPath.Parse($"{cp}/system/receptionist")) + .ToImmutableHashSet(); + centralClient = system.ActorOf( + ClusterClient.Props(ClusterClientSettings.Create(system).WithInitialContacts(contacts)), + CentralClusterClientName); + } + + var nodeComm = system.ActorOf( + NodeCommunicationActor.Props(centralClient), + 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 + // make flipping the flag require a node restart before central could reach it. + ClusterClientReceptionist.Get(system).RegisterService(nodeComm); + registry.Register(nodeComm); + // Continuous-historization recorder — gated on ContinuousHistorization:Enabled AND the // gateway-backed IHistorianValueWriter + the durable IHistorizationOutbox being registered // (the Host registers both ONLY when historization is enabled and the ServerHistorian gateway @@ -398,7 +439,11 @@ public static class ServiceCollectionExtensions registry.Register(peerProbes); var driverHost = system.ActorOf( - DriverHostActor.Props(dbFactory, roleInfo.LocalNode, ackRouter: null, + // 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. + DriverHostActor.Props(dbFactory, roleInfo.LocalNode, + ackRouter: useClusterClient ? nodeComm : null, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, dependencyMux: mux, opcUaPublishActor: publishActor, @@ -430,6 +475,7 @@ public sealed class DbHealthProbeActorKey { } public sealed class HistorianAdapterActorKey { } public sealed class DependencyMuxActorKey { } public sealed class OpcUaPublishActorKey { } +public sealed class NodeCommunicationActorKey { } /// Marker key for the per-node ContinuousHistorizationRecorder (spawned only when /// ContinuousHistorization:Enabled=true and the gateway value-writer + outbox are registered). diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs new file mode 100644 index 00000000..427106b3 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs @@ -0,0 +1,62 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// Pins the two mesh comm-actor paths against a really booted host. +/// +/// +/// +/// /user/central-communication and /user/node-communication are the wire +/// contract of the central↔node boundary: central sends to a path string, and the node +/// registers under one. Nothing else in the codebase asserts they agree. Renaming either +/// actor would compile, pass every unit test, deploy — and deliver nothing, with no error +/// on either side, because a ClusterClient send to an unregistered path is simply dropped. +/// +/// +/// Booting the real host is the point. A test that spawned the actors itself would pin the +/// constant against itself and prove nothing about the wiring in +/// WithOtOpcUaControlPlaneSingletons / WithOtOpcUaRuntimeActors. +/// +/// +public class MeshCommActorPathTests +{ + private static async Task AssertActorExistsAsync(ActorSystem system, string path) + { + var identity = await system.ActorSelection(path) + .Ask(new Identify(path), TimeSpan.FromSeconds(10)); + + identity.Subject.ShouldNotBeNull( + $"{path} is the ClusterClient wire contract for the central↔node boundary; nothing else " + + "pins it, and a mismatch delivers nothing without erroring on either side"); + } + + [Fact] + public async Task Both_mesh_comm_actors_exist_at_their_contract_paths() + { + await using var harness = await TwoNodeClusterHarness.StartAsync(); + + // Harness nodes carry admin+driver, so BOTH comm actors live on each of them. + await AssertActorExistsAsync(harness.NodeASystem, MeshPaths.CentralCommunication); + await AssertActorExistsAsync(harness.NodeASystem, MeshPaths.NodeCommunication); + await AssertActorExistsAsync(harness.NodeBSystem, MeshPaths.CentralCommunication); + await AssertActorExistsAsync(harness.NodeBSystem, MeshPaths.NodeCommunication); + } + + // NOT TESTED HERE: that the comm actors are registered per node rather than as cluster + // singletons. Two candidate discriminators were tried and BOTH were invalid: + // + // * "the path resolves on both nodes" — a ClusterSingletonManager is also created on every + // node in the role, so this passes under either shape. + // * "no /singleton child exists" — measured against the KNOWN singleton /user/config-publish, + // that child is null there too, so Akka.Hosting does not lay singletons out this way and + // the assertion proves nothing. + // + // A sabotage that re-registered CentralCommunicationActor via WithSingleton left both green, + // which is what exposed them. Rather than ship an assertion I could not make fail, the property + // is left to MeshClusterClientBoundaryTests: a ClusterClient send only arrives if the target is + // actually registered with the receptionist, which is the thing that matters. +}