From 6531ec19847b92f9357f8aa4d1804b9f5c151cd9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 01:53:20 -0400 Subject: [PATCH] feat(mesh): one ClusterClient per Cluster in CentralCommunicationActor (Phase 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After the fleet splits into one Akka mesh per application Cluster, a ClusterClientReceptionist serves only its own mesh — so the single fleet-wide client's SendToAll reached only the mesh whose receptionist answered, leaving every other cluster silently on its old configuration. Central now holds one ClusterClient per ClusterId and fans SendToAll across all of them. LoadContactsFromDb selects ClusterId and groups the receptionist contacts per cluster (keeping the per-row TryParse guard and the enabled/non-maintenance filter); ContactsLoaded carries ContactsByCluster; HandleContactsLoaded diffs per cluster (rebuild changed, stop+drop vanished, warn-don't-create on empty); RebuildClient builds a per-cluster client under a clusterId-derived actor name. IMeshClusterClientFactory.Create gained a clusterId parameter so the per-cluster clients get distinct, diagnosable names. ApplyAck handling and the DPS branch are unchanged; the deploy path stays payload-free. Tests: focused unit coverage of the grouping + fan-out (one-client-per-cluster, fan-out SendToAll to every cluster client) via the recording factory double, plus the real two-mesh boundary test extended to central + two separate site meshes proving one dispatch reaches both and a client scoped to one cluster never crosses into another. Red-before-green verified: crippling the fan-out to a single client fails the reaches-both-meshes assertion. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../CentralCommunicationActor.cs | 180 ++++++++++++------ .../IMeshClusterClientFactory.cs | 26 ++- .../CentralCommunicationActorTests.cs | 56 +++++- .../MeshClusterClientFactoryTests.cs | 21 +- .../MeshClusterClientBoundaryTests.cs | 150 ++++++++++++++- 5 files changed, 358 insertions(+), 75 deletions(-) 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. ///