dohertj2/lmxopcua
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mesh): NodeCommunicationActor — inbound commands to the node-local EventStream

Browse Source 
Phase 2 Task 4. Node-side end of the boundary, registered with the receptionist
per node rather than as a singleton so central's contact rotation reaches
whichever node answers.

Inbound commands are re-emitted on the node's EventStream, NOT back onto their
DistributedPubSub topic. DPS is mesh-wide and central SendToAll-s to every
node's comm actor, so a DPS republish would deliver N copies of every command
to every subscriber. The EventStream is node-local by construction. It also
avoids a registration handshake with actors spawned later: ScriptedAlarmHostActor
is a CHILD of DriverHostActor and has no registry key, so there is no ref to
hand in at wiring time.

Each inbound type is listed explicitly rather than caught by a ReceiveAny, so an
unknown type dead-letters loudly instead of being republished blind onto a
stream where every subscriber ignores it.

Outbound is ApplyAck only, and it uses Send rather than SendToAll -- the deploy
coordinator is one singleton behind central's proxy, so SendToAll would deliver
one ack per central node and the coordinator would count this node twice.
Notably there is NO Ask across this boundary in Phase 2: every migrated command
is fire-and-forget, which is why this actor is far smaller than the sister
project's equivalent and needs no sender preservation at all.

Sabotage-verified twice: deleting the AlarmCommand handler and flipping Send to
SendToAll each redden exactly their own test.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-22 12:12:12 -04:00
parent d5b5cb6ede
commit 915beec11a
2 changed files with 224 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs
+94
View File
@@ -0,0 +1,94 @@
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Event;
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;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication;
/// <summary>
/// Node-side end of the central↔node command boundary (per-cluster mesh Phase 2), registered
/// with the <c>ClusterClientReceptionist</c> at <see cref="MeshPaths.NodeCommunication"/>.
/// </summary>
/// <remarks>
/// <para>
/// <b>Registered per node, NOT as a cluster singleton</b> — central's ClusterClient rotates
/// across contact points and must find a live comm actor at whichever node answers.
/// </para>
/// <para>
/// <b>Inbound commands are re-emitted on the node's <see cref="EventStream"/>, not on their
/// DistributedPubSub topic.</b> DPS is mesh-wide, and central <c>SendToAll</c>s to every
/// node's comm actor — so a DPS re-publish would deliver N copies of every command to every
/// subscriber, where N is the node count. The EventStream is node-local by construction,
/// which is exactly the needed semantics. It also avoids a registration handshake with
/// actors spawned later: <c>ScriptedAlarmHostActor</c> is a <i>child of</i>
/// <c>DriverHostActor</c> and has no registry key, so there is no ref to hand in at wiring
/// time.
/// </para>
/// <para>
/// <b>Outbound is <see cref="ApplyAck"/> only.</b> There is no Ask across this boundary in
/// Phase 2 — every migrated command is fire-and-forget, and the deploy ack returns as an
/// independent message the coordinator already subscribes for. Nothing here needs sender
/// preservation, which is why this actor is far smaller than the sister project's.
/// </para>
/// </remarks>
public sealed class NodeCommunicationActor : ReceiveActor
{
private readonly IActorRef? _centralClient;
private readonly ILoggingAdapter _log = Context.GetLogger();
/// <summary>Creates the props for this actor.</summary>
/// <param name="centralClient">
/// ClusterClient dialling central's receptionist, or <see langword="null"/> under
/// <c>MeshTransport:Mode = Dps</c>, where acks still go out over DistributedPubSub from
/// <c>DriverHostActor</c> and this actor never sees them.
/// </param>
/// <returns>The props.</returns>
public static Props Props(IActorRef? centralClient) =>
Akka.Actor.Props.Create(() => new NodeCommunicationActor(centralClient));
/// <summary>Initializes a new instance of the <see cref="NodeCommunicationActor"/> class.</summary>
/// <param name="centralClient">ClusterClient dialling central, or <see langword="null"/>.</param>
public NodeCommunicationActor(IActorRef? centralClient)
{
_centralClient = centralClient;
// 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
// blind onto the node's EventStream, where it would be silently ignored by every subscriber.
Receive<DispatchDeployment>(PublishLocally);
Receive<RestartDriver>(PublishLocally);
Receive<ReconnectDriver>(PublishLocally);
Receive<AlarmCommand>(PublishLocally);
// Outbound to central.
Receive<ApplyAck>(HandleApplyAck);
}
private void PublishLocally(object msg)
{
_log.Debug("Mesh: received {MessageType} from central; publishing node-locally", msg.GetType().Name);
Context.System.EventStream.Publish(msg);
}
private void HandleApplyAck(ApplyAck ack)
{
if (_centralClient is null)
{
_log.Warning(
"No central ClusterClient — 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",
ack.DeploymentId);
return;
}
// Send, not SendToAll: the deploy coordinator is a single singleton, and the central comm
// actor forwards to its proxy. SendToAll would deliver one ack per central node, and the
// coordinator would count the same node twice.
_centralClient.Tell(new ClusterClient.Send(MeshPaths.CentralCommunication, ack));
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/NodeCommunicationActorTests.cs
+130
View File
@@ -0,0 +1,130 @@
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Shouldly;
using Xunit;
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.Runtime.Communication;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication;
public class NodeCommunicationActorTests : RuntimeActorTestBase
{
private static DispatchDeployment NewDispatch() => new(
new DeploymentId(Guid.NewGuid()),
new RevisionHash("rev-1"),
new CorrelationId(Guid.NewGuid()));
[Fact]
public void DispatchDeployment_from_central_lands_on_the_node_local_event_stream()
{
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
var dispatch = NewDispatch();
actor.Tell(dispatch);
subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5)).ShouldBe(dispatch);
}
[Fact]
public void A_command_is_published_exactly_once()
{
// The reason inbound commands do NOT go back onto their DPS topic: central SendToAll-s to
// every node's comm actor, so a mesh-wide republish would deliver N copies to every
// subscriber. One publish per inbound command is the whole point.
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(DispatchDeployment));
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
actor.Tell(NewDispatch());
subscriber.ExpectMsg<DispatchDeployment>(TimeSpan.FromSeconds(5));
subscriber.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public void RestartDriver_from_central_lands_on_the_event_stream()
{
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(RestartDriver));
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
var msg = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid());
actor.Tell(msg);
subscriber.ExpectMsg<RestartDriver>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
}
[Fact]
public void ReconnectDriver_from_central_lands_on_the_event_stream()
{
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(ReconnectDriver));
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
var msg = new ReconnectDriver("C1", "driver-1", "alice", Guid.NewGuid());
actor.Tell(msg);
subscriber.ExpectMsg<ReconnectDriver>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
}
[Fact]
public void AlarmCommand_from_central_lands_on_the_event_stream()
{
// ScriptedAlarmHostActor is a CHILD of DriverHostActor with no registry key, so there is no
// ref to hand this actor at wiring time — the EventStream is what makes the alarm-command
// leg possible without a registration handshake.
var subscriber = CreateTestProbe();
Sys.EventStream.Subscribe(subscriber.Ref, typeof(AlarmCommand));
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
var msg = new AlarmCommand("alarm-1", "Acknowledge", "alice", "comment", null);
actor.Tell(msg);
subscriber.ExpectMsg<AlarmCommand>(TimeSpan.FromSeconds(5)).ShouldBe(msg);
}
[Fact]
public void ApplyAck_goes_to_central_as_a_Send_not_a_SendToAll()
{
// Send, not SendToAll: the deploy coordinator is one singleton behind the central comm
// actor's proxy. SendToAll would deliver one ack per central node and the coordinator would
// count this node twice.
var client = CreateTestProbe();
var actor = Sys.ActorOf(NodeCommunicationActor.Props(client.Ref));
var ack = new ApplyAck(
new DeploymentId(Guid.NewGuid()),
new NodeId("host-a:4053"),
ApplyAckOutcome.Applied,
null,
new CorrelationId(Guid.NewGuid()));
actor.Tell(ack);
var sent = client.ExpectMsg<ClusterClient.Send>(TimeSpan.FromSeconds(5));
sent.Path.ShouldBe(MeshPaths.CentralCommunication);
sent.Message.ShouldBe(ack);
}
[Fact]
public void An_ack_with_no_central_client_is_dropped_with_a_warning()
{
var actor = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null));
var ack = new ApplyAck(
new DeploymentId(Guid.NewGuid()),
new NodeId("host-a:4053"),
ApplyAckOutcome.Applied,
null,
new CorrelationId(Guid.NewGuid()));
EventFilter.Warning(contains: "dropping ApplyAck").ExpectOne(
TimeSpan.FromSeconds(5), () => actor.Tell(ack));
}
}