diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 5c13c198..7a842744 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -86,7 +86,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private readonly IDbContextFactory _dbFactory; private readonly CommonsNodeId _localNode; - private readonly IActorRef? _coordinatorOverride; + private readonly IActorRef? _ackRouter; private readonly IDriverFactory _driverFactory; /// Builds the per-driver-instance attached to each @@ -328,7 +328,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Creates props for a DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. /// The local cluster node identifier. - /// Optional coordinator actor reference for deployment coordination. + /// Where ApplyAcks are sent. A direct coordinator handle in tests; under + /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays + /// across the boundary. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -364,7 +366,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers public static Props Props( IDbContextFactory dbFactory, CommonsNodeId localNode, - IActorRef? coordinator = null, + IActorRef? ackRouter = null, IDriverFactory? driverFactory = null, IReadOnlySet? localRoles = null, IActorRef? dependencyMux = null, @@ -387,7 +389,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // the wrong dependency at runtime. New parameters go LAST in all three places — this // signature, the constructor's, and this call — and nothing else moves. Akka.Actor.Props.Create(() => new DriverHostActor( - dbFactory, localNode, coordinator, driverFactory, localRoles, dependencyMux, opcUaPublishActor, + dbFactory, localNode, ackRouter, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost)); @@ -395,7 +397,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. /// The local cluster node identifier. - /// Optional coordinator actor reference for deployment coordination. + /// Where ApplyAcks are sent. A direct coordinator handle in tests; under + /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays + /// across the boundary. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -426,7 +430,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers public DriverHostActor( IDbContextFactory dbFactory, CommonsNodeId localNode, - IActorRef? coordinator, + IActorRef? ackRouter, IDriverFactory? driverFactory = null, IReadOnlySet? localRoles = null, IActorRef? dependencyMux = null, @@ -449,7 +453,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _replicationPeerHost = replicationPeerHost; _dbFactory = dbFactory; _localNode = localNode; - _coordinatorOverride = coordinator; + _ackRouter = ackRouter; _driverFactory = driverFactory ?? NullDriverFactory.Instance; _invokerFactory = invokerFactory ?? NullDriverCapabilityInvokerFactory.Instance; _driverMemberCountProvider = driverMemberCountProvider; @@ -483,6 +487,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // uses so only the Primary services operator writes (the secondary keeps state warm for failover). _mediator.Tell( new Subscribe(ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RedundancyStateTopic, Self)); + + // Per-cluster mesh Phase 2 dark switch. Under MeshTransport:Mode = ClusterClient these same + // commands arrive from NodeCommunicationActor on the node-local EventStream instead of the + // mesh-wide DPS topic. Subscribing to BOTH unconditionally is deliberate: exactly one of the + // two ever publishes, so the transport flag lives entirely on the publishing side and the + // switch can be flipped (or reverted) without touching any subscriber. + Context.System.EventStream.Subscribe(Self, typeof(DispatchDeployment)); + Context.System.EventStream.Subscribe(Self, typeof(RestartDriver)); + Context.System.EventStream.Subscribe(Self, typeof(ReconnectDriver)); + // Spawn the VirtualTag host BEFORE Bootstrap so the bootstrap-restore path (which routes // through PushDesiredSubscriptions and Tells ApplyVirtualTags) has a live host to target. SpawnVirtualTagHost(); @@ -2410,15 +2424,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private void SendAck(DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation) { var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); - if (_coordinatorOverride is not null) + if (_ackRouter is not null) { - _coordinatorOverride.Tell(ack); + // Either a direct coordinator handle (tests) or, under + // MeshTransport:Mode = ClusterClient, the NodeCommunicationActor that relays this + // across the boundary. Renamed from _coordinatorOverride in per-cluster mesh Phase 2: + // under ClusterClient mode this ref is emphatically NOT a coordinator, and the old + // name would send the next reader looking for one. + _ackRouter.Tell(ack); } else { - // No direct coordinator handle — publish on the dedicated ACK topic. The coordinator - // singleton subscribes there in PreStart so the ACK reaches whichever admin node hosts - // it without an actor-path lookup. + // No router — publish on the dedicated ACK topic. The coordinator singleton subscribes + // there in PreStart so the ACK reaches whichever admin node hosts it without an + // actor-path lookup. DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentAcksTopic, ack)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index 5c465063..7b538c66 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -528,6 +528,15 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor // RedundancyStateChanged and cache this node's role — OnEngineEmission gates the cluster-wide // alerts publish to the Primary so a 2-node single-cluster deploy doesn't double-emit transitions. _mediator.Tell(new Subscribe(OpcUaPublishActor.RedundancyStateTopic, Self)); + + // Per-cluster mesh Phase 2 dark switch. An AdminUI-originated ack/shelve arrives from + // NodeCommunicationActor on the node-local EventStream under + // MeshTransport:Mode = ClusterClient; the DPS subscribe above still carries the node-LOCAL + // publisher (OtOpcUaServerHostedService, the OPC UA Part 9 method calls), which never + // crosses the boundary and stays on DPS in both modes. Subscribing to both is safe: exactly + // one publisher feeds each path. + Context.System.EventStream.Subscribe(Self, typeof(AlarmCommand)); + base.PreStart(); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 9cdbeddb..0cb262c7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -398,7 +398,7 @@ public static class ServiceCollectionExtensions registry.Register(peerProbes); var driverHost = system.ActorOf( - DriverHostActor.Props(dbFactory, roleInfo.LocalNode, coordinator: null, + DriverHostActor.Props(dbFactory, roleInfo.LocalNode, ackRouter: null, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, dependencyMux: mux, opcUaPublishActor: publishActor, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/EventStreamCommandDeliveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/EventStreamCommandDeliveryTests.cs new file mode 100644 index 00000000..c9c66be1 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/EventStreamCommandDeliveryTests.cs @@ -0,0 +1,161 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using Serilog; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication; + +/// +/// Proves the Phase 2 node-side leg joins up: a command handed to +/// reaches the real , +/// not just a test probe. +/// +/// +/// +/// The unit tests either side of this seam both pass with the wiring broken. +/// NodeCommunicationActorTests asserts a probe subscribed to the EventStream receives +/// the message; the driver-host tests drive the actor by direct Tell. Neither notices +/// if DriverHostActor.PreStart never subscribes. That gap is why this file exists. +/// +/// +/// Ordering matters, and it bit this test first time out. ActorOf returns +/// before PreStart has run, and an EventStream.Publish with no subscriber is +/// dropped silently — there is no buffering and no dead-letter. Relaying immediately after +/// spawning therefore fails intermittently. The warm-up below is not ceremony: it is the +/// only available proof that PreStart has completed. (In production the driver host +/// is spawned at startup, long before any deployment, so the same race does not arise.) +/// +/// +public class EventStreamCommandDeliveryTests : RuntimeActorTestBase +{ + private static DispatchDeployment NewDispatch() => new( + new DeploymentId(Guid.NewGuid()), + new RevisionHash("rev-does-not-exist"), + new CorrelationId(Guid.NewGuid())); + + [Fact] + public void A_dispatch_relayed_by_the_comm_actor_reaches_the_real_driver_host() + { + var ackRouter = CreateTestProbe(); + var driverHost = Sys.ActorOf( + DriverHostActor.Props(NewInMemoryDbFactory(), new NodeId("127.0.0.1:4053"), ackRouter: ackRouter.Ref), + "driver-host"); + var comm = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null), "node-communication"); + + // Warm-up: a direct Tell whose ack proves PreStart has completed, and with it the + // EventStream subscription. Without this the relay below races the subscription. + var warmup = NewDispatch(); + driverHost.Tell(warmup); + ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(warmup.DeploymentId); + + // Now the real assertion — relayed exactly as central would send it. The comm actor only + // publishes node-locally, so the host must be subscribed for this to arrive at all. + var relayed = NewDispatch(); + comm.Tell(relayed); + + // The deployment id does not exist, so the host answers Failed. That is the stronger + // assertion, not a weaker one: an ack at all proves the command was received and handled, + // and only PreStart's EventStream subscription can have delivered it. + ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(relayed.DeploymentId); + } + + [Fact] + public void CONTROL_the_driver_host_acks_a_direct_tell() + { + // Falsifiability control. If the test above fails, this one says which half broke: green + // here means the host is healthy and the EventStream relay is at fault; red here means the + // host would not have acked either way and the relay is not implicated. + var ackRouter = CreateTestProbe(); + var driverHost = Sys.ActorOf( + DriverHostActor.Props(NewInMemoryDbFactory(), new NodeId("127.0.0.1:4053"), ackRouter: ackRouter.Ref), + "driver-host-control"); + + var dispatch = NewDispatch(); + driverHost.Tell(dispatch); + + ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(dispatch.DeploymentId); + } + +} + +/// +/// The alarm leg of the same wiring proof, in its own ActorSystem because the shared +/// harness pins loglevel = WARNING and the only +/// observable signal ScriptedAlarmHostActor gives for an unowned alarm is a Debug line. +/// +public class AlarmCommandEventStreamDeliveryTests : TestKit +{ + private static string DebugLevelHocon => @" +akka { + loglevel = ""DEBUG"" + extensions = [ + ""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools"" + ] + actor { + provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"" + } + remote.dot-netty.tcp { + hostname = ""127.0.0.1"" + port = 0 + } + cluster { + seed-nodes = [] + roles = [""driver""] + min-nr-of-members = 1 + run-coordinated-shutdown-when-down = off + } +}"; + + public AlarmCommandEventStreamDeliveryTests() + : base(DebugLevelHocon) + { + var cluster = Akka.Cluster.Cluster.Get(Sys); + cluster.Join(cluster.SelfAddress); + AwaitCondition( + () => cluster.State.Members.Any(m => m.Status == Akka.Cluster.MemberStatus.Up), + TimeSpan.FromSeconds(5)); + } + + [Fact] + public void An_alarm_command_relayed_by_the_comm_actor_reaches_the_real_scripted_alarm_host() + { + // Needed for the same reason as the deploy leg: the comm actor's unit test only proves a + // probe receives it, and the alarm-host tests drive the actor by direct Tell. Neither + // notices a missing PreStart subscription. + var upstream = new DependencyMuxTagUpstreamSource(); + var logger = new LoggerConfiguration().CreateLogger(); + var engine = new ScriptedAlarmEngine( + upstream, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger); + + var host = Sys.ActorOf( + ScriptedAlarmHostActor.Props( + CreateTestProbe().Ref, CreateTestProbe().Ref, upstream, engine, new NodeId("node-A")), + "scripted-alarm-host"); + var comm = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null), "node-communication"); + + var cmd = new AlarmCommand("alarm-unowned", "Acknowledge", "alice", "c", null); + + // Warm-up, OUTSIDE the real assertion: a direct Tell whose debug line proves PreStart has + // completed and the EventStream subscription is registered. An earlier draft did the direct + // Tell INSIDE the assertion window alongside the relay, which made the test vacuous — the + // direct Tell alone produced the log the assertion was waiting for. + EventFilter.Debug(contains: "ignoring AlarmCommand") + .ExpectOne(TimeSpan.FromSeconds(15), () => host.Tell(cmd)); + + // The real assertion: relayed ONLY through the comm actor, which publishes node-locally. + // The host owns no alarms, so it logs and stops — still a positive delivery signal, because + // that line is only reached if the message arrived. + EventFilter.Debug(contains: "ignoring AlarmCommand") + .ExpectOne(TimeSpan.FromSeconds(15), () => comm.Tell(cmd)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs index dc03c043..dbd1323e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -41,7 +41,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var cache = new StubArtifactCache(cached); var actor = Sys.ActorOf(DriverHostActor.Props( - new ThrowingDbFactory(), TestNode, coordinator: null, + new ThrowingDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); @@ -70,7 +70,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var publishProbe = CreateTestProbe(); Sys.ActorOf(DriverHostActor.Props( - new ThrowingDbFactory(), TestNode, coordinator: null, + new ThrowingDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, opcUaPublishActor: publishProbe.Ref, deploymentArtifactCache: new StubArtifactCache(cached))); @@ -85,7 +85,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase { // The pre-cache behaviour, unchanged. A cache miss must not invent a configuration. var actor = Sys.ActorOf(DriverHostActor.Props( - new ThrowingDbFactory(), TestNode, coordinator: null, + new ThrowingDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: new StubArtifactCache(current: null))); @@ -100,7 +100,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase { // Admin-only graphs and older harnesses pass null; behaviour is identical to a cache miss. var actor = Sys.ActorOf(DriverHostActor.Props( - new ThrowingDbFactory(), TestNode, coordinator: null, + new ThrowingDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" })); var snapshot = AskDiagnostics(actor); @@ -118,7 +118,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase DeploymentId.NewId().ToString(), CachedRev.Value, EmptyArtifact(), DateTimeOffset.UtcNow)); var actor = Sys.ActorOf(DriverHostActor.Props( - NewInMemoryDbFactory(), TestNode, coordinator: null, + NewInMemoryDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); @@ -140,7 +140,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase deploymentId.ToString(), CachedRev.Value, EmptyArtifact(), DateTimeOffset.UtcNow)); var actor = Sys.ActorOf(DriverHostActor.Props( - db, TestNode, coordinator: null, + db, TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); @@ -156,7 +156,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase { // A cache fault must leave the node exactly where it would have been without a cache. var actor = Sys.ActorOf(DriverHostActor.Props( - new ThrowingDbFactory(), TestNode, coordinator: null, + new ThrowingDbFactory(), TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: new ThrowingReadCache())); @@ -183,7 +183,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var cache = new StubArtifactCache(current: null); var actor = Sys.ActorOf(DriverHostActor.Props( - db, TestNode, coordinator: null, + db, TestNode, ackRouter: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs index 67f8cbbf..0e0ec0ce 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRoleViewTests.cs @@ -183,7 +183,7 @@ public sealed class DriverHostActorRoleViewTests : RuntimeActorTestBase Sys.ActorOf(DriverHostActor.Props( NewInMemoryDbFactory(), TestNode, - coordinator: null, + ackRouter: null, driverMemberCountProvider: () => driverMemberCount, redundancyRoleView: view, replicationPeerHost: PeerHost)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs index 48495808..be6dd343 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs @@ -139,7 +139,7 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase var hostActor = Sys.ActorOf(DriverHostActor.Props( dbFactory: NewInMemoryDbFactory(), localNode: ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId.Parse("host-1"), - coordinator: hostProbe.Ref, + ackRouter: hostProbe.Ref, dependencyMux: mux)); // Tell the host an AttributeValuePublished — it should fan out to the mux + subscriber.