From 2a2e54f0833793d262f9a864d5e52e8418acc3ca Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 22 Jul 2026 16:50:27 -0400 Subject: [PATCH] feat(mesh): route deploy, driver-control and alarm commands through the mesh router Phase 2 Task 7. The five central->node publish sites -- ConfigPublishCoordinator's deploy dispatch, and AdminOperationsActor's restart, reconnect, acknowledge and shelve -- now hand a MeshCommand to the transport router instead of telling the DistributedPubSub mediator directly. Neither actor knows which transport is in force any more. _meshRouter is NULLABLE with a mediator fallback, and that is deliberate: about a dozen existing coordinator and admin-operations tests construct these actors without a router, and they must keep exercising the real DPS path rather than a silently-disabled one. The fallback is not a permanent seam -- Phase 6 deletes it with the DPS branch and the single mesh. Wiring note: the comm actor's WithActors block moved ABOVE the singleton registrations so both singletons can resolve it with registry.Get, which is the ordering-by-registration idiom AdminOperationsActor already uses for the coordinator. An earlier version used ActorSelection.ResolveOne().GetAwaiter() .GetResult() inside the props factory -- that blocks inside actor construction and races the very spawn it waits for. Sabotage-verified by inverting the branch so the mediator always wins: all six new tests go red, including the fallback test (which proves it is asserting the mediator path rather than passing by accident). Suites after: ControlPlane 118/118, Runtime 450/450, Host.IntegrationTests 196/202 -- sole failure AbCip_Green_AgainstSim, the fixture baseline that fails on master too. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../AdminOperations/AdminOperationsActor.cs | 48 ++++++- .../Coordinators/ConfigPublishCoordinator.cs | 32 ++++- .../ServiceCollectionExtensions.cs | 74 +++++----- .../Communication/MeshRouterDispatchTests.cs | 134 ++++++++++++++++++ 4 files changed, 243 insertions(+), 45 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshRouterDispatchTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs index 6766c520..4b000bf4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs @@ -4,6 +4,7 @@ using Akka.Event; using Microsoft.EntityFrameworkCore; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; 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.Types; using ZB.MOM.WW.OtOpcUa.Configuration; @@ -26,6 +27,13 @@ public sealed class AdminOperationsActor : ReceiveActor private readonly IActorRef _coordinator; private readonly IReadOnlyDictionary _probesByType; private readonly TagConfigValidationMode _tagConfigValidationMode; + + /// + /// Where central→node commands are handed off (per-cluster mesh Phase 2), or + /// to publish on the mediator directly. + /// + private readonly IActorRef? _meshRouter; + private readonly ILoggingAdapter _log = Context.GetLogger(); /// Creates actor props for the admin operations actor. @@ -33,29 +41,38 @@ public sealed class AdminOperationsActor : ReceiveActor /// Reference to the deployment coordinator actor. /// Driver probes registered in DI; keyed by DriverType (case-insensitive). /// Deploy-gate TagConfig strictness mode (Warn default | Error opt-in). + /// + /// Central comm actor commands are handed to, or to publish on the + /// DistributedPubSub mediator directly (the pre-Phase-2 behaviour). + /// /// Props configured to create an AdminOperationsActor. public static Props Props( IDbContextFactory dbFactory, IActorRef coordinator, IEnumerable probes, - TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn) => - Akka.Actor.Props.Create(() => new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode)); + TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn, + IActorRef? meshRouter = null) => + Akka.Actor.Props.Create(() => + new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode, meshRouter)); /// Initializes a new instance of the AdminOperationsActor. /// Factory for creating config database contexts. /// Reference to the deployment coordinator actor. /// Driver probes registered in DI; keyed by DriverType (case-insensitive). /// Deploy-gate TagConfig strictness mode (Warn default | Error opt-in). + /// Central comm actor, or for the mediator. public AdminOperationsActor( IDbContextFactory dbFactory, IActorRef coordinator, IEnumerable probes, - TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn) + TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn, + IActorRef? meshRouter = null) { _dbFactory = dbFactory; _coordinator = coordinator; _probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase); _tagConfigValidationMode = tagConfigValidationMode; + _meshRouter = meshRouter; ReceiveAsync(HandleStartDeploymentAsync); ReceiveAsync(HandleTestDriverConnectAsync); @@ -65,6 +82,23 @@ public sealed class AdminOperationsActor : ReceiveActor Receive(HandleShelveAlarm); } + /// + /// Hands a central→node command to the mesh transport router, or publishes it directly when + /// no router is wired. + /// + /// + /// The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh Phase 2) + /// on exactly the pre-Phase-2 behaviour. It is not a permanent seam: Phase 6 deletes it along + /// with the DistributedPubSub branch and the single mesh. + /// + /// The legacy DistributedPubSub topic. + /// The command. + private void Dispatch(string topic, object message) + { + if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(topic, message)); + else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(topic, message)); + } + /// /// AdminUI Acknowledge path. Maps the control-plane command to a /// (Operation = "Acknowledge") and publishes it onto the @@ -85,7 +119,7 @@ public sealed class AdminOperationsActor : ReceiveActor Comment: msg.Comment, UnshelveAtUtc: null); - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(AlarmCommandsTopic.Name, alarmCmd)); + Dispatch(AlarmCommandsTopic.Name, alarmCmd); _log.Info("AdminOps: Acknowledge published for alarm {AlarmId} by {User}", msg.AlarmId, msg.User); replyTo.Tell(new AcknowledgeAlarmResult(true, null, msg.CorrelationId)); @@ -132,7 +166,7 @@ public sealed class AdminOperationsActor : ReceiveActor Comment: msg.Comment, UnshelveAtUtc: unshelveAt); - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(AlarmCommandsTopic.Name, alarmCmd)); + Dispatch(AlarmCommandsTopic.Name, alarmCmd); _log.Info("AdminOps: {Operation} published for alarm {AlarmId} by {User}", operation, msg.AlarmId, msg.User); replyTo.Tell(new ShelveAlarmResult(true, null, msg.CorrelationId)); @@ -394,7 +428,7 @@ public sealed class AdminOperationsActor : ReceiveActor { // Broadcast to every DriverHostActor on every node via the driver-control DPS topic. // Only the host that owns the instance will act; others ignore it (id not found in _children). - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DriverControlTopic.Name, msg)); + Dispatch(DriverControlTopic.Name, msg); await using var db = await _dbFactory.CreateDbContextAsync(); db.ConfigEdits.Add(new ConfigEdit @@ -424,7 +458,7 @@ public sealed class AdminOperationsActor : ReceiveActor try { // Broadcast to every DriverHostActor; only the one owning the instance reacts. - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DriverControlTopic.Name, msg)); + Dispatch(DriverControlTopic.Name, msg); await using var db = await _dbFactory.CreateDbContextAsync(); db.ConfigEdits.Add(new ConfigEdit diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index d955dc58..86787656 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -3,6 +3,7 @@ using Akka.Cluster.Tools.PublishSubscribe; using Akka.Event; using Microsoft.EntityFrameworkCore; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; @@ -60,6 +61,17 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers private readonly IDbContextFactory _dbFactory; private readonly TimeSpan _applyDeadline; + + /// + /// Where central→node commands are handed off (per-cluster mesh Phase 2), or + /// to publish on the mediator directly. + /// + /// + /// Null in tests and on any node wired before Phase 2, where the fallback reproduces the + /// pre-Phase-2 behaviour exactly. The fallback is not a permanent seam — Phase 6 deletes it + /// along with the DistributedPubSub branch and the single mesh. + /// + private readonly IActorRef? _meshRouter; private readonly ILoggingAdapter _log = Context.GetLogger(); // NodeId equality here is case-insensitive (by Value) to match the case-insensitive ClusterId/ // NodeId scoping in DeploymentArtifact.ResolveClusterScope — so an ack from a node whose address @@ -78,22 +90,31 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// The database context factory for accessing configuration data. /// The timeout for waiting for apply acknowledgments from cluster nodes. /// The Akka.NET used to spawn this actor. + /// + /// Central comm actor the dispatch is handed to, or to publish on the + /// DistributedPubSub mediator directly (the pre-Phase-2 behaviour). + /// public static Props Props( IDbContextFactory dbFactory, - TimeSpan? applyDeadline = null) => - Akka.Actor.Props.Create(() => new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline)); + TimeSpan? applyDeadline = null, + IActorRef? meshRouter = null) => + Akka.Actor.Props.Create(() => + new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter)); /// /// Initializes a new instance of the class. /// /// The database context factory for persisting deployment state. /// The timeout for waiting for per-node apply acknowledgments. + /// Central comm actor, or for the mediator. public ConfigPublishCoordinator( IDbContextFactory dbFactory, - TimeSpan applyDeadline) + TimeSpan applyDeadline, + IActorRef? meshRouter = null) { _dbFactory = dbFactory; _applyDeadline = applyDeadline; + _meshRouter = meshRouter; Receive(HandleDispatch); Receive(HandleAck); Receive(HandleDeadline); @@ -172,7 +193,10 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers db.SaveChanges(); } - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentsTopic, msg)); + // Per-cluster mesh Phase 2: hand off to the transport router rather than publishing + // directly, so this actor no longer knows which transport is in force. + if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(DeploymentsTopic, msg)); + else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentsTopic, msg)); Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(msg.DeploymentId), _applyDeadline); if (_expectedAcks.Count == 0) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 73c7e6d7..86e1561b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -53,12 +53,49 @@ public static class ServiceCollectionExtensions var singletonOptions = new ClusterSingletonOptions { Role = AdminRole }; var proxyOptions = new ClusterSingletonOptions { Role = AdminRole }; + // 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); + }); + builder.WithSingleton( ConfigPublishSingletonName, (system, registry, resolver) => { var dbFactory = resolver.GetService>(); - return ConfigPublishCoordinator.Props(dbFactory); + // Registered above, so it is already in the registry — the same + // ordering-by-registration the AdminOperations singleton relies on to resolve the + // coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks + // inside a props factory and races the spawn it is waiting for. + var meshRouter = registry.Get(); + return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: meshRouter); }, singletonOptions); @@ -74,7 +111,8 @@ public static class ServiceCollectionExtensions var mode = Enum.TryParse( resolver.GetService()?["Deployment:TagConfigValidationMode"], ignoreCase: true, out var m) ? m : TagConfigValidationMode.Warn; - return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode); + var meshRouter = registry.Get(); + return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter); }, singletonOptions); @@ -117,38 +155,6 @@ 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; } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshRouterDispatchTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshRouterDispatchTests.cs new file mode 100644 index 00000000..fdb17d6d --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshRouterDispatchTests.cs @@ -0,0 +1,134 @@ +using Akka.Actor; +using Akka.Cluster.Tools.PublishSubscribe; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; + +/// +/// Proves the five central→node publish sites go through the transport router when one is +/// wired, and fall back to the mediator when it is not. +/// +/// +/// The fallback is what keeps the ~12 pre-existing coordinator and admin-operations tests +/// meaningful: they construct these actors without a router and must still exercise the real +/// DistributedPubSub path, not a silently-disabled one. +/// +public class MeshRouterDispatchTests : ControlPlaneActorTestBase +{ + private const string DeploymentsTopic = "deployments"; + + [Fact] + public void Deploy_dispatch_goes_to_the_router_when_one_is_wired() + { + var router = CreateTestProbe(); + var dbFactory = NewInMemoryDbFactory(); + var coordinator = Sys.ActorOf( + ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: router.Ref)); + + var deploymentId = new DeploymentId(Guid.NewGuid()); + coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment( + deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId())); + + var cmd = router.ExpectMsg(TimeSpan.FromSeconds(5)); + cmd.Topic.ShouldBe(DeploymentsTopic); + cmd.Message.ShouldBeOfType(); + } + + [Fact] + public void Deploy_dispatch_falls_back_to_the_mediator_when_no_router_is_wired() + { + var subscriber = CreateTestProbe(); + DistributedPubSub.Get(Sys).Mediator.Tell( + new Subscribe(DeploymentsTopic, subscriber.Ref), subscriber.Ref); + subscriber.ExpectMsg(TimeSpan.FromSeconds(5)); + + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(NewInMemoryDbFactory())); + + var deploymentId = new DeploymentId(Guid.NewGuid()); + coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment( + deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId())); + + subscriber.ExpectMsg(TimeSpan.FromSeconds(5)); + } + + [Fact] + public void Restart_driver_goes_to_the_router_on_the_driver_control_topic() + { + var router = CreateTestProbe(); + var admin = Sys.ActorOf(AdminOperationsActor.Props( + NewInMemoryDbFactory(), + CreateTestProbe().Ref, + Enumerable.Empty(), + TagConfigValidationMode.Warn, + router.Ref)); + + admin.Tell(new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid())); + + var cmd = router.ExpectMsg(TimeSpan.FromSeconds(5)); + cmd.Topic.ShouldBe(DriverControlTopic.Name); + cmd.Message.ShouldBeOfType(); + } + + [Fact] + public void Reconnect_driver_goes_to_the_router_on_the_driver_control_topic() + { + var router = CreateTestProbe(); + var admin = Sys.ActorOf(AdminOperationsActor.Props( + NewInMemoryDbFactory(), + CreateTestProbe().Ref, + Enumerable.Empty(), + TagConfigValidationMode.Warn, + router.Ref)); + + admin.Tell(new ReconnectDriver("C1", "driver-1", "alice", Guid.NewGuid())); + + var cmd = router.ExpectMsg(TimeSpan.FromSeconds(5)); + cmd.Topic.ShouldBe(DriverControlTopic.Name); + cmd.Message.ShouldBeOfType(); + } + + [Fact] + public void Alarm_acknowledge_goes_to_the_router_on_the_alarm_commands_topic() + { + var router = CreateTestProbe(); + var admin = Sys.ActorOf(AdminOperationsActor.Props( + NewInMemoryDbFactory(), + CreateTestProbe().Ref, + Enumerable.Empty(), + TagConfigValidationMode.Warn, + router.Ref)); + + admin.Tell(new AcknowledgeAlarmCommand("alarm-1", "alice", "c", CorrelationId.NewId())); + + var cmd = router.ExpectMsg(TimeSpan.FromSeconds(5)); + cmd.Topic.ShouldBe(AlarmCommandsTopic.Name); + cmd.Message.ShouldBeOfType(); + } + + [Fact] + public void Alarm_shelve_goes_to_the_router_on_the_alarm_commands_topic() + { + var router = CreateTestProbe(); + var admin = Sys.ActorOf(AdminOperationsActor.Props( + NewInMemoryDbFactory(), + CreateTestProbe().Ref, + Enumerable.Empty(), + TagConfigValidationMode.Warn, + router.Ref)); + + admin.Tell(new ShelveAlarmCommand("alarm-1", "alice", ShelveKind.OneShot, null, "c", CorrelationId.NewId())); + + var cmd = router.ExpectMsg(TimeSpan.FromSeconds(5)); + cmd.Topic.ShouldBe(AlarmCommandsTopic.Name); + cmd.Message.ShouldBeOfType(); + } +}