diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
index 333087e6..f400ea17 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
@@ -25,8 +25,11 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
/// rotation would fail silently against the other.
///
///
-/// Exactly one ClusterClient, fleet-wide. See — this is
-/// the single most important thing to understand before changing this actor.
+/// One ClusterClient per application Cluster (Phase 6). The fleet is now one
+/// Akka mesh per cluster, so a single fleet-wide client's SendToAll would reach only
+/// the one mesh whose receptionist answered. This actor therefore holds a client per cluster
+/// and fans SendToAll across all of them. See — this is the
+/// single most important thing to understand before changing this actor.
///
///
public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
@@ -40,8 +43,12 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly bool _useClusterClient;
- private IActorRef? _client;
- private ImmutableHashSet _currentContacts = ImmutableHashSet.Empty;
+ // One client per application Cluster, keyed by ClusterId. A ClusterClient reaches only the mesh
+ // whose receptionist it dialled, and after Phase 6 every cluster is its own mesh — so central must
+ // hold a client per cluster and fan SendToAll across all of them.
+ private readonly Dictionary Contacts)> _clients =
+ new();
+
private CancellationTokenSource _lifecycle = new();
/// Gets the timer scheduler driving the periodic contact refresh.
@@ -121,6 +128,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
///
protected override void PostStop()
{
+ foreach (var (client, _) in _clients.Values) Context.Stop(client);
+ _clients.Clear();
+
_lifecycle.Cancel();
_lifecycle.Dispose();
}
@@ -133,7 +143,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
return;
}
- if (_client is null)
+ if (_clients.Count == 0)
{
// App-level drop, matching the sister project's decision and this repo's buffer-size = 0.
// The command is NOT queued and will NOT be retried: a deploy fails at the coordinator's
@@ -141,19 +151,21 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
// command arriving minutes late against a config the DB already recorded as failed.
_log.Warning(
"No mesh ClusterClient — dropping {MessageType}. The last contact refresh produced "
- + "no usable contact point from the enabled, non-maintenance ClusterNode rows; the "
- + "command is not buffered and will not be retried",
+ + "no usable contact point from the enabled, non-maintenance ClusterNode rows in any "
+ + "cluster; the command is not buffered and will not be retried",
cmd.Message.GetType().Name);
return;
}
- // SendToAll, NEVER Send. Today's DPS publish reaches EVERY DriverHostActor — there is no
- // ClusterId or node filter on the node side at all; scoping happens later, inside the
- // artifact (DeploymentArtifact.ResolveClusterScope). ClusterClient.Send delivers to exactly
- // ONE registered actor, which would deploy to a single node of the fleet while every other
- // node silently kept its old configuration — and the deployment would still seal green if
- // the ack set happened to be satisfied.
- _client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, cmd.Message));
+ // SendToAll, NEVER Send, fanned across EVERY cluster's client. Today's DPS publish reaches
+ // EVERY DriverHostActor — there is no ClusterId or node filter on the node side at all; scoping
+ // happens later, inside the artifact (DeploymentArtifact.ResolveClusterScope). Each ClusterClient
+ // reaches only the one mesh it dialled, so central must fan across all of them to preserve the
+ // fleet-wide reach DPS gives. ClusterClient.Send delivers to exactly ONE registered actor, which
+ // would deploy to a single node while every other node silently kept its old configuration — and
+ // the deployment would still seal green if the ack set happened to be satisfied.
+ foreach (var (client, _) in _clients.Values)
+ client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, cmd.Message));
}
private void HandleApplyAck(ApplyAck ack)
@@ -198,11 +210,14 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
var rows = await db.ClusterNodes
.AsNoTracking()
.Where(n => n.Enabled && !n.MaintenanceMode)
- .Select(n => new { n.NodeId, n.Host, n.AkkaPort })
+ .Select(n => new { n.NodeId, n.ClusterId, n.Host, n.AkkaPort })
.ToListAsync(ct)
.ConfigureAwait(false);
- var contacts = new List(rows.Count);
+ // Grouped BY ClusterId: central now dials one ClusterClient per application Cluster, so the
+ // contact set must be partitioned the same way. A row's contacts join only its own cluster's
+ // client.
+ var byCluster = new Dictionary>();
var malformed = new List();
foreach (var row in rows)
{
@@ -212,11 +227,21 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
// not abort the whole refresh and leave the contact set half-built — the sister
// project shipped that bug and its regression test deliberately orders the bad row
// first.
- if (ActorPath.TryParse(address, out _)) contacts.Add(address);
- else malformed.Add($"{row.NodeId} -> {address}");
+ if (ActorPath.TryParse(address, out _))
+ {
+ if (!byCluster.TryGetValue(row.ClusterId, out var list))
+ byCluster[row.ClusterId] = list = new List();
+ list.Add(address);
+ }
+ else
+ {
+ malformed.Add($"{row.NodeId} ({row.ClusterId}) -> {address}");
+ }
}
- return new ContactsLoaded(contacts, malformed);
+ var contactsByCluster = byCluster.ToDictionary(
+ kv => kv.Key, kv => (IReadOnlyList)kv.Value);
+ return new ContactsLoaded(contactsByCluster, malformed);
}, ct).PipeTo(self);
}
@@ -229,70 +254,108 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
+ "refresh (other nodes are unaffected). Check the row's Host and AkkaPort", bad);
}
- var contacts = msg.Contacts.ToImmutableHashSet();
+ var byCluster = msg.ContactsByCluster.ToDictionary(
+ kv => kv.Key, kv => kv.Value.ToImmutableHashSet());
- if (contacts.IsEmpty)
+ // Clusters that dropped out of the DB entirely, or whose only rows were malformed this refresh:
+ // stop and forget their clients so a vanished cluster stops receiving. A cluster whose rows all
+ // went malformed lands here via the empty-set branch.
+ var gone = _clients.Keys
+ .Where(id => !byCluster.TryGetValue(id, out var c) || c.IsEmpty)
+ .ToList();
+ foreach (var clusterId in gone)
{
- _log.Warning(
- "No usable ClusterNode contact points; the mesh ClusterClient was not created and "
- + "every central→node command will be dropped until a refresh finds one");
- return;
+ _log.Info(
+ "Cluster {ClusterId} no longer yields any usable contact point; stopping its "
+ + "ClusterClient", clusterId);
+ Context.Stop(_clients[clusterId].Client);
+ _clients.Remove(clusterId);
}
- if (_client is not null && _currentContacts.SetEquals(contacts)) return;
+ var anyUsable = false;
+ foreach (var (clusterId, contacts) in byCluster)
+ {
+ if (contacts.IsEmpty)
+ {
+ // Only-malformed rows for this cluster this refresh. Warn per cluster; its client (if
+ // any) was already stopped above.
+ _log.Warning(
+ "Cluster {ClusterId} has no usable ClusterNode contact points; no ClusterClient "
+ + "was created for it and its nodes' commands will be dropped until a refresh finds "
+ + "one", clusterId);
+ continue;
+ }
- RebuildClient(contacts);
+ anyUsable = true;
+
+ if (_clients.TryGetValue(clusterId, out var existing) && existing.Contacts.SetEquals(contacts))
+ continue;
+
+ RebuildClient(clusterId, contacts);
+ }
+
+ if (!anyUsable && _clients.Count == 0)
+ {
+ _log.Warning(
+ "No usable ClusterNode contact points in any cluster; no mesh ClusterClient was "
+ + "created and every central→node command will be dropped until a refresh finds one");
+ }
}
- /// Stops any existing client and creates a replacement for the new contact set.
+ /// Stops a cluster's existing client and creates a replacement for its new contact set.
///
///
- /// TODO(Phase 6): one client per application Cluster. Today this actor
- /// creates exactly one client whose contacts are the whole fleet, and that is not
- /// a compromise — it is the only correct shape while the fleet is a single Akka mesh.
+ /// One client per application Cluster (delivered in Phase 6). This actor now
+ /// holds one ClusterClient per cluster and fans
+ /// SendToAll across all of them. Once the fleet is split into one Akka mesh per
+ /// cluster, a ClusterClientReceptionist serves only its own mesh, so a single
+ /// client's SendToAll would reach the one mesh whose receptionist answered and leave
+ /// every other cluster silently on its old configuration. Fanning across a client per
+ /// cluster restores the fleet-wide reach the pre-split DPS broadcast gave.
///
///
- /// A ClusterClientReceptionist serves its entire cluster, so
- /// SendToAll reaches every registered node-comm actor in the mesh regardless of
- /// which node's address was used as the contact point. One client per Cluster would
- /// therefore fan each command out once per cluster — N× duplicate
- /// DispatchDeployment and N× duplicate ApplyAck. Phase 6 splits the meshes,
- /// at which point per-cluster clients become both correct and necessary.
+ /// Why this is not N× duplicate dispatch: each client dials only its own cluster's
+ /// receptionists, so a command is delivered once per mesh, not once per cluster into
+ /// every mesh. (While the fleet was a single mesh, the same code would have duplicated —
+ /// which is exactly why the fan-out could not land until the meshes were partitioned.)
///
///
/// Corollary worth stating because it is easy to assume otherwise: the contact set
- /// does not scope delivery. Excluding a maintenance-mode node from the contacts does
- /// not stop it receiving commands — it still receives them, exactly as it does under
- /// today's DPS broadcast. The filter is about which receptionists are worth dialling
- /// (a node in maintenance may be switched off), not about who gets the message.
+ /// does not scope delivery within a mesh. Excluding a maintenance-mode node from a
+ /// cluster's contacts does not stop it receiving commands — its receptionist still delivers
+ /// to every registered node-comm actor in that mesh. The filter is about which receptionists
+ /// are worth dialling (a node in maintenance may be switched off), not about who gets
+ /// the message.
///
///
- /// The receptionist addresses to dial.
- private void RebuildClient(ImmutableHashSet contacts)
+ /// The application Cluster whose client is being (re)built.
+ /// The receptionist addresses to dial for that cluster.
+ private void RebuildClient(string clusterId, ImmutableHashSet contacts)
{
- if (_client is not null)
+ if (_clients.TryGetValue(clusterId, out var existing))
{
- _log.Info("Mesh contact set changed; replacing the ClusterClient");
- Context.Stop(_client);
+ _log.Info("Contact set for cluster {ClusterId} changed; replacing its ClusterClient", clusterId);
+ Context.Stop(existing.Client);
- // Cleared now: if the create below throws, a stale ref would route commands into a
+ // Removed now: if the create below throws, a stale ref would route commands into a
// stopping actor and they would vanish without a warning.
- _client = null;
- _currentContacts = ImmutableHashSet.Empty;
+ _clients.Remove(clusterId);
}
try
{
var paths = contacts.Select(ActorPath.Parse).ToImmutableHashSet();
- _client = _clientFactory.Create(Context.System, paths);
- _currentContacts = contacts;
- _log.Info("Mesh ClusterClient created with {Count} contact point(s)", contacts.Count);
+ var client = _clientFactory.Create(Context.System, clusterId, paths);
+ _clients[clusterId] = (client, contacts);
+ _log.Info(
+ "Mesh ClusterClient for cluster {ClusterId} created with {Count} contact point(s)",
+ clusterId, contacts.Count);
}
catch (Exception ex)
{
_log.Error(ex,
- "Failed to create the mesh ClusterClient; central→node commands are dropped until "
- + "the next contact refresh succeeds");
+ "Failed to create the mesh ClusterClient for cluster {ClusterId}; its nodes' commands "
+ + "are dropped until the next contact refresh succeeds", clusterId);
}
}
@@ -300,9 +363,12 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
public sealed record RefreshContacts;
/// Result of a contact-point load, piped back from the DB read.
- /// Receptionist addresses that parsed cleanly.
- /// Rows that did not, described for the log.
+ ///
+ /// Receptionist addresses that parsed cleanly, grouped by the owning application
+ /// Cluster's id. One ClusterClient is built per key.
+ ///
+ /// Rows that did not parse, described for the log.
public sealed record ContactsLoaded(
- IReadOnlyList Contacts,
+ IReadOnlyDictionary> ContactsByCluster,
IReadOnlyList Malformed);
}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs
index 5ea57434..83b2b2ee 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs
@@ -13,9 +13,14 @@ public interface IMeshClusterClientFactory
{
/// Creates a client for the given receptionist contact points.
/// The actor system to spawn the client in.
+ ///
+ /// The application Cluster these contacts belong to. Central now creates one client per
+ /// cluster (Phase 6), so the id is woven into the client's actor name to keep the per-cluster
+ /// clients distinct and diagnosable.
+ ///
/// Receptionist actor paths, one per reachable node.
/// The new client's .
- IActorRef Create(ActorSystem system, ImmutableHashSet contacts);
+ IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet contacts);
}
///
@@ -35,13 +40,28 @@ public sealed class DefaultMeshClusterClientFactory : IMeshClusterClientFactory
private long _generation;
///
- public IActorRef Create(ActorSystem system, ImmutableHashSet contacts)
+ public IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet contacts)
{
ArgumentNullException.ThrowIfNull(system);
+ ArgumentException.ThrowIfNullOrEmpty(clusterId);
ArgumentNullException.ThrowIfNull(contacts);
var settings = ClusterClientSettings.Create(system).WithInitialContacts(contacts);
- var name = $"mesh-client-{Interlocked.Increment(ref _generation)}";
+ var name = $"mesh-client-{SanitizeForActorName(clusterId)}-{Interlocked.Increment(ref _generation)}";
return system.ActorOf(ClusterClient.Props(settings), name);
}
+
+ ///
+ /// Reduces a cluster id to characters safe in an Akka actor name. The id is operator-authored
+ /// (e.g. a GUID or a slug), so it may carry characters Akka rejects; anything outside
+ /// [A-Za-z0-9-] becomes -. The trailing generation counter still guarantees
+ /// uniqueness, so a lossy sanitisation cannot cause a name collision.
+ ///
+ /// The raw cluster id.
+ /// An actor-name-safe rendering of the id.
+ private static string SanitizeForActorName(string clusterId)
+ {
+ var chars = clusterId.Select(c => char.IsAsciiLetterOrDigit(c) || c == '-' ? c : '-');
+ return new string(chars.ToArray());
+ }
}
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
index d2d0fc7a..97e3f28f 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
@@ -31,13 +31,20 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
public List> Calls { get; } = [];
+ public List ClusterIds { get; } = [];
+
public List Probes { get; } = [];
- public IActorRef Create(ActorSystem system, ImmutableHashSet contacts)
+ /// Probe created for a given cluster id, for per-cluster fan-out assertions.
+ public Dictionary ProbeByCluster { get; } = [];
+
+ public IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet contacts)
{
Calls.Add(contacts);
+ ClusterIds.Add(clusterId);
var probe = _newProbe();
Probes.Add(probe);
+ ProbeByCluster[clusterId] = probe;
return probe.Ref;
}
}
@@ -138,23 +145,54 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
}
[Fact]
- public void Exactly_one_client_is_created_for_a_multi_cluster_fleet()
+ public void One_client_per_cluster_is_created_for_a_multi_cluster_fleet()
{
- // A receptionist serves its whole mesh, so one client per Cluster would fan each command out
- // once per cluster while the fleet is still a single mesh — N x duplicate deployments.
+ // THE Phase 6 inversion. Each application Cluster is now its own Akka mesh, and a ClusterClient
+ // reaches only the mesh whose receptionist it dialled. Central must therefore build one client
+ // per cluster, each dialling only that cluster's nodes.
var factory = NewInMemoryDbFactory();
SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
SeedCluster(factory, "C2", ("host-b:4053", "host-b", 4053, true, false));
var clients = new RecordingClientFactory(() => CreateTestProbe());
Spawn(factory, ClusterClientMode(), clients);
- AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5));
+ AwaitCondition(() => clients.Calls.Count == 2, TimeSpan.FromSeconds(5));
- // ...and that one client dials BOTH clusters' nodes.
- clients.Calls[0].Count.ShouldBe(2);
- // Give a second refresh a chance to (wrongly) create another.
+ // Give a second refresh a chance to (wrongly) create more.
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
- clients.Calls.Count.ShouldBe(1);
+ clients.Calls.Count.ShouldBe(2);
+ clients.ClusterIds.ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true);
+
+ // Each client dials ONLY its own cluster's node.
+ clients.Calls[clients.ClusterIds.IndexOf("C1")].Single().ToString().ShouldContain("host-a");
+ clients.Calls[clients.ClusterIds.IndexOf("C2")].Single().ToString().ShouldContain("host-b");
+ }
+
+ [Fact]
+ public void A_command_fans_out_a_send_to_all_to_every_cluster_client()
+ {
+ // The other half of the inversion: one MeshCommand must reach EVERY cluster's mesh, so central
+ // fans SendToAll across all per-cluster clients. A single Send/one-client shape would leave
+ // every cluster but one silently on its old configuration.
+ var factory = NewInMemoryDbFactory();
+ SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
+ SeedCluster(factory, "C2", ("host-b:4053", "host-b", 4053, true, false));
+ var clients = new RecordingClientFactory(() => CreateTestProbe());
+
+ var actor = Spawn(factory, ClusterClientMode(), clients);
+ AwaitCondition(() => clients.Probes.Count == 2, TimeSpan.FromSeconds(5));
+
+ var dispatch = new DispatchDeployment(
+ new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));
+ actor.Tell(new MeshCommand(DeploymentsTopic, dispatch));
+
+ foreach (var (clusterId, probe) in clients.ProbeByCluster)
+ {
+ var sent = probe.ExpectMsg(
+ TimeSpan.FromSeconds(5), $"cluster {clusterId} must receive the fanned SendToAll");
+ sent.Path.ShouldBe(MeshPaths.NodeCommunication);
+ sent.Message.ShouldBe(dispatch);
+ }
}
[Fact]
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs
index 911142b7..3c188719 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs
@@ -37,13 +37,13 @@ public class MeshClusterClientFactoryTests : TestKit
{
var factory = new DefaultMeshClusterClientFactory();
- var first = factory.Create(Sys, UnreachableContact);
+ var first = factory.Create(Sys, "C1", UnreachableContact);
Sys.Stop(first);
// Stop is asynchronous: `first`'s name is still reserved right here. Without the generation
// suffix this throws InvalidActorNameException — which is precisely what a contact-set
// change would do in production, where stop-then-recreate happens in one handler.
- var second = factory.Create(Sys, UnreachableContact);
+ var second = factory.Create(Sys, "C1", UnreachableContact);
second.Path.Name.ShouldNotBe(first.Path.Name);
}
@@ -54,12 +54,27 @@ public class MeshClusterClientFactoryTests : TestKit
var factory = new DefaultMeshClusterClientFactory();
var names = Enumerable.Range(0, 5)
- .Select(_ => factory.Create(Sys, UnreachableContact).Path.Name)
+ .Select(_ => factory.Create(Sys, "C1", UnreachableContact).Path.Name)
.ToList();
names.Distinct().Count().ShouldBe(5);
}
+ [Fact]
+ public void The_cluster_id_is_woven_into_the_client_actor_name()
+ {
+ // Per-cluster clients must be distinguishable in logs and the actor tree, and — more
+ // importantly — two clusters' clients created on one system must never collide on a name.
+ var factory = new DefaultMeshClusterClientFactory();
+
+ var a = factory.Create(Sys, "site-a", UnreachableContact);
+ var b = factory.Create(Sys, "site-b", UnreachableContact);
+
+ a.Path.Name.ShouldContain("site-a");
+ b.Path.Name.ShouldContain("site-b");
+ a.Path.Name.ShouldNotBe(b.Path.Name);
+ }
+
// Contact-set propagation is deliberately NOT asserted here: reading initial contacts back
// off a created ClusterClient is not possible, and asserting ClusterClientSettings directly
// would test Akka rather than this factory. It is covered where it can actually fail — the
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs
index 18592757..267dd3ce 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs
@@ -3,12 +3,16 @@ using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.Client;
using Akka.Configuration;
+using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
+using ZB.MOM.WW.OtOpcUa.Cluster;
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.Types;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
using AkkaCluster = Akka.Cluster.Cluster;
@@ -86,12 +90,12 @@ public sealed class MeshClusterClientBoundaryTests : IAsyncLifetime
ClusterClientReceptionist.Get(_central).RegisterService(centralComm);
// Node: the REAL comm actor, dialling central through the REAL production client factory.
- var nodeToCentral = _clientFactory.Create(_node, ReceptionistOf(_central));
+ var nodeToCentral = _clientFactory.Create(_node, "central", ReceptionistOf(_central));
_nodeComm = _node.ActorOf(
NodeCommunicationActor.Props(nodeToCentral), MeshPaths.NodeCommunicationName);
ClusterClientReceptionist.Get(_node).RegisterService(_nodeComm);
- _centralClient = _clientFactory.Create(_central, ReceptionistOf(_node));
+ _centralClient = _clientFactory.Create(_central, "site-a", ReceptionistOf(_node));
}
///
@@ -179,7 +183,7 @@ public sealed class MeshClusterClientBoundaryTests : IAsyncLifetime
var comm = deaf.ActorOf(NodeCommunicationActor.Props(null), MeshPaths.NodeCommunicationName);
comm.ShouldNotBeNull();
- var client = _clientFactory.Create(_central, ReceptionistOf(deaf));
+ var client = _clientFactory.Create(_central, "deaf", ReceptionistOf(deaf));
var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid());
var deadline = DateTime.UtcNow + Silence;
@@ -199,6 +203,146 @@ public sealed class MeshClusterClientBoundaryTests : IAsyncLifetime
}
}
+ [Fact]
+ public async Task One_dispatch_through_central_reaches_a_node_in_every_cluster_mesh()
+ {
+ // THE Phase 6 assertion at the transport level, driving the REAL CentralCommunicationActor.
+ // Two genuinely separate site meshes (site-A = _node, site-B created here), each its own
+ // single-member cluster with its own receptionist. Central must build ONE ClusterClient per
+ // cluster and fan SendToAll across them, so a single MeshCommand reaches BOTH meshes. A single
+ // fleet-wide client would reach only the one mesh whose receptionist answered — the exact
+ // failure this test is red-before-green against.
+ var siteB = StartSelfSeededSystem(withReceptionist: true);
+ try
+ {
+ await AwaitSingleMemberUpAsync(siteB);
+ RegisterNodeComm(siteB);
+
+ var inboxA = NewInbox();
+ _node.EventStream.Subscribe(_node.ActorOf(CaptureActor.Props(inboxA)), typeof(RestartDriver));
+ var inboxB = NewInbox();
+ siteB.EventStream.Subscribe(siteB.ActorOf(CaptureActor.Props(inboxB)), typeof(RestartDriver));
+
+ var dbFactory = SeedTwoClusterDb(SiteAddress(_node), SiteAddress(siteB));
+ var actor = _central.ActorOf(CentralCommunicationActor.Props(
+ dbFactory, ClusterClientOptions(), new DefaultMeshClusterClientFactory(), () => null));
+
+ var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid());
+ var deadline = DateTime.UtcNow + Arrival;
+ while (DateTime.UtcNow < deadline && !(inboxA.Task.IsCompleted && inboxB.Task.IsCompleted))
+ {
+ actor.Tell(new MeshCommand("deployments", command));
+ await Task.Delay(250, TestContext.Current.CancellationToken);
+ }
+
+ inboxA.Task.IsCompleted.ShouldBeTrue(
+ "site-A's mesh must receive the dispatch through its own per-cluster ClusterClient");
+ inboxB.Task.IsCompleted.ShouldBeTrue(
+ "site-B's mesh must receive the SAME dispatch through its own per-cluster ClusterClient; "
+ + "a single fleet-wide client would have reached only one of the two meshes");
+ }
+ finally
+ {
+ await ShutdownAsync(siteB);
+ }
+ }
+
+ [Fact]
+ public async Task A_client_built_for_one_cluster_does_not_deliver_into_another_clusters_mesh()
+ {
+ // The partition proof for the fan-out: a client dialling ONLY site-A's contacts must land in
+ // site-A's mesh and NEVER cross into site-B's. This is why the per-cluster fan-out is correct
+ // rather than accidentally N x broadcasting — each client is scoped to its own mesh.
+ var siteB = StartSelfSeededSystem(withReceptionist: true);
+ try
+ {
+ await AwaitSingleMemberUpAsync(siteB);
+ RegisterNodeComm(siteB);
+
+ var inboxA = NewInbox();
+ _node.EventStream.Subscribe(_node.ActorOf(CaptureActor.Props(inboxA)), typeof(RestartDriver));
+ var inboxB = NewInbox();
+ siteB.EventStream.Subscribe(siteB.ActorOf(CaptureActor.Props(inboxB)), typeof(RestartDriver));
+
+ var clientA = _clientFactory.Create(_central, "site-a", ReceptionistOf(_node));
+ var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid());
+
+ // Keep sending for the full Silence window so any erroneous cross-delivery into site-B has
+ // every chance to arrive — a pass means the boundary is shut, not that the clock ran out.
+ var deadline = DateTime.UtcNow + Silence;
+ while (DateTime.UtcNow < deadline)
+ {
+ clientA.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, command));
+ await Task.Delay(250, TestContext.Current.CancellationToken);
+ }
+
+ inboxA.Task.IsCompleted.ShouldBeTrue("site-A's client must deliver into site-A's mesh");
+ inboxB.Task.IsCompleted.ShouldBeFalse(
+ "a client built for site-A's contacts must NOT deliver into site-B's mesh — the "
+ + "per-cluster fan-out must stay partitioned, one client per mesh");
+ }
+ finally
+ {
+ await ShutdownAsync(siteB);
+ }
+ }
+
+ /// Registers a real node-comm actor with a system's receptionist so central can reach it.
+ /// The site system to register into.
+ private static void RegisterNodeComm(ActorSystem system)
+ {
+ var comm = system.ActorOf(NodeCommunicationActor.Props(null), MeshPaths.NodeCommunicationName);
+ ClusterClientReceptionist.Get(system).RegisterService(comm);
+ }
+
+ private static (string Host, int Port) SiteAddress(ActorSystem system)
+ {
+ var addr = AkkaCluster.Get(system).SelfAddress;
+ return (addr.Host!, addr.Port!.Value);
+ }
+
+ private static MeshTransportOptions ClusterClientOptions() => new()
+ {
+ Mode = MeshTransportOptions.ModeClusterClient,
+ CentralContactPoints = ["akka.tcp://otopcua@127.0.0.1:4053"],
+ ContactRefreshSeconds = 1,
+ };
+
+ private static IDbContextFactory SeedTwoClusterDb(
+ (string Host, int Port) siteA, (string Host, int Port) siteB)
+ {
+ var factory = new InMemoryConfigDbFactory(Guid.NewGuid().ToString("N"));
+ using var db = factory.CreateDbContext();
+ db.ClusterNodes.Add(NodeRow("SITE-A", "node-a", siteA));
+ db.ClusterNodes.Add(NodeRow("SITE-B", "node-b", siteB));
+ db.SaveChanges();
+ return factory;
+ }
+
+ private static ClusterNode NodeRow(string clusterId, string nodeId, (string Host, int Port) addr) => new()
+ {
+ NodeId = nodeId,
+ ClusterId = clusterId,
+ Host = addr.Host,
+ AkkaPort = addr.Port,
+ ApplicationUri = $"urn:{nodeId}",
+ Enabled = true,
+ MaintenanceMode = false,
+ CreatedBy = "test",
+ };
+
+ private sealed class InMemoryConfigDbFactory(string dbName)
+ : IDbContextFactory
+ {
+ public OtOpcUaConfigDbContext CreateDbContext()
+ {
+ var opts = new DbContextOptionsBuilder()
+ .UseInMemoryDatabase(dbName)
+ .Options;
+ return new OtOpcUaConfigDbContext(opts);
+ }
+ }
+
///
/// Starts a single-member cluster on a dynamic port, optionally without the receptionist.
///