feat(mesh): node subscribers accept EventStream commands; _coordinatorOverride -> _ackRouter

Phase 2 Task 5. DriverHostActor and ScriptedAlarmHostActor now subscribe to the node-local EventStream alongside their existing DPS topic subscriptions, so the transport flag lives ENTIRELY on the publishing side -- exactly one of the two ever publishes, and the switch can be flipped or reverted without touching a single subscriber. Note the alarm case is not symmetric: the DPS subscribe there still carries the node-LOCAL publisher (OtOpcUaServerHostedService's OPC UA Part 9 method calls), which never crosses the boundary and stays on DPS in both modes. Only the central AdminUI leg moves. Renamed DriverHostActor._coordinatorOverride to _ackRouter. Under ClusterClient mode that ref is the NodeCommunicationActor, emphatically not a coordinator, and the old name would send the next reader looking for one. Adds two wiring tests, because the unit tests either side of this seam BOTH pass with the subscription deleted -- the comm actor's test asserts a probe receives the message, and the host tests drive the actor by direct Tell. Neither notices a missing PreStart subscribe. Sabotage-verified: deleting either subscription reddens exactly its own wiring test and nothing else. Three things went wrong while writing those tests, all worth keeping: - The first version raced. ActorOf returns before PreStart runs, and an EventStream.Publish with no subscriber is dropped silently -- no buffering, no dead letter. Fixed with a warm-up whose ack proves PreStart completed; the comment says why it is not ceremony. - The alarm version was VACUOUS: it told the host directly INSIDE the assertion window alongside the relay, so the direct Tell alone produced the log the assertion waited for. Split into warm-up (outside) and relay (inside). - The alarm test needs its own ActorSystem: the shared harness pins loglevel = WARNING and the only signal an unowned alarm gives is a Debug line. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-22 12:19:29 -04:00
parent 915beec11a
commit 5cb0e72166
7 changed files with 212 additions and 23 deletions
@@ -86,7 +86,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly CommonsNodeId _localNode;
-    private readonly IActorRef? _coordinatorOverride;
+    private readonly IActorRef? _ackRouter;
    private readonly IDriverFactory _driverFactory;

    /// <summary>Builds the per-driver-instance <see cref="IDriverCapabilityInvoker"/> attached to each
@@ -328,7 +328,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
    /// <summary>Creates props for a DriverHostActor with the specified dependencies.</summary>
    /// <param name="dbFactory">Database context factory for configuration database access.</param>
    /// <param name="localNode">The local cluster node identifier.</param>
-    /// <param name="coordinator">Optional coordinator actor reference for deployment coordination.</param>
+    /// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under
+    /// <c>MeshTransport:Mode = ClusterClient</c> the node's <c>NodeCommunicationActor</c>, which relays
+    /// across the boundary. Null publishes on the <c>deployment-acks</c> DPS topic.</param>
    /// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param>
    /// <param name="localRoles">Optional set of roles assigned to the local node.</param>
    /// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param>
@@ -364,7 +366,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        CommonsNodeId localNode,
-        IActorRef? coordinator = null,
+        IActorRef? ackRouter = null,
        IDriverFactory? driverFactory = null,
        IReadOnlySet<string>? 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
    /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
    /// <param name="dbFactory">Database context factory for configuration database access.</param>
    /// <param name="localNode">The local cluster node identifier.</param>
-    /// <param name="coordinator">Optional coordinator actor reference for deployment coordination.</param>
+    /// <param name="ackRouter">Where ApplyAcks are sent. A direct coordinator handle in tests; under
+    /// <c>MeshTransport:Mode = ClusterClient</c> the node's <c>NodeCommunicationActor</c>, which relays
+    /// across the boundary. Null publishes on the <c>deployment-acks</c> DPS topic.</param>
    /// <param name="driverFactory">Optional driver factory; defaults to null factory if not provided.</param>
    /// <param name="localRoles">Optional set of roles assigned to the local node.</param>
    /// <param name="dependencyMux">Optional actor reference for dependency multiplexing.</param>
@@ -426,7 +430,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
    public DriverHostActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        CommonsNodeId localNode,
-        IActorRef? coordinator,
+        IActorRef? ackRouter,
        IDriverFactory? driverFactory = null,
        IReadOnlySet<string>? 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));
        }
    }
@@ -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();
    }

@@ -398,7 +398,7 @@ public static class ServiceCollectionExtensions
            registry.Register<PeerProbeSupervisorKey>(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,
@@ -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;
+
+/// <summary>
+///     Proves the Phase 2 node-side leg joins up: a command handed to
+///     <see cref="NodeCommunicationActor"/> reaches the <b>real</b> <see cref="DriverHostActor"/>,
+///     not just a test probe.
+/// </summary>
+/// <remarks>
+///     <para>
+///         The unit tests either side of this seam both pass with the wiring broken.
+///         <c>NodeCommunicationActorTests</c> asserts a probe subscribed to the EventStream receives
+///         the message; the driver-host tests drive the actor by direct <c>Tell</c>. Neither notices
+///         if <c>DriverHostActor.PreStart</c> never subscribes. That gap is why this file exists.
+///     </para>
+///     <para>
+///         <b>Ordering matters, and it bit this test first time out.</b> <c>ActorOf</c> returns
+///         before <c>PreStart</c> has run, and an <c>EventStream.Publish</c> 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 <c>PreStart</c> has completed. (In production the driver host
+///         is spawned at startup, long before any deployment, so the same race does not arise.)
+///     </para>
+/// </remarks>
+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<ApplyAck>(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<ApplyAck>(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<ApplyAck>(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(dispatch.DeploymentId);
+    }
+
+}
+
+/// <summary>
+///     The alarm leg of the same wiring proof, in its own ActorSystem because the shared
+///     <see cref="RuntimeActorTestBase"/> harness pins <c>loglevel = WARNING</c> and the only
+///     observable signal <c>ScriptedAlarmHostActor</c> gives for an unowned alarm is a Debug line.
+/// </summary>
+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));
+    }
+}
@@ -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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "driver" },
            deploymentArtifactCache: cache));

@@ -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));
@@ -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.