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

feat(mesh): register comm actors with the receptionist; pin the wire-contract paths

Browse Source 
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<IActorRef?> +
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
This commit is contained in:
Joseph Doherty
2026-07-22 16:42:07 -04:00
parent 95e4f97529
commit 16d598560a
3 changed files with 147 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs
+38
View File
@@ -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<IDbContextFactory<OtOpcUaConfigDbContext>>();
var options = resolver.GetService<IOptions<MeshTransportOptions>>().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<ConfigPublishCoordinatorKey>(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<CentralCommunicationActorKey>(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 { }
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs
+47 -1
View File
@@ -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";
/// <summary>
/// Registers shared runtime services. Currently binds <see cref="IAlarmHistorianSink"/>
@@ -319,6 +326,40 @@ public static class ServiceCollectionExtensions
var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName);
registry.Register<DependencyMuxActorKey>(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<IOptions<MeshTransportOptions>>().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<NodeCommunicationActorKey>(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<PeerProbeSupervisorKey>(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 { }
/// <summary>Marker key for the per-node ContinuousHistorizationRecorder (spawned only when
/// <c>ContinuousHistorization:Enabled=true</c> and the gateway value-writer + outbox are registered).</summary>
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs
+62
View File
@@ -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;
/// <summary>
/// Pins the two mesh comm-actor paths against a <b>really booted</b> host.
/// </summary>
/// <remarks>
/// <para>
/// <c>/user/central-communication</c> and <c>/user/node-communication</c> 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.
/// </para>
/// <para>
/// 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
/// <c>WithOtOpcUaControlPlaneSingletons</c> / <c>WithOtOpcUaRuntimeActors</c>.
/// </para>
/// </remarks>
public class MeshCommActorPathTests
{
private static async Task AssertActorExistsAsync(ActorSystem system, string path)
{
var identity = await system.ActorSelection(path)
.Ask<ActorIdentity>(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.
}