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 316a1c4a..c3443d67 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs @@ -1,6 +1,4 @@ -using System.Collections.Immutable; using Akka.Actor; -using Akka.Cluster.Tools.Client; using Akka.Cluster.Tools.PublishSubscribe; using Akka.Event; using Grpc.Core; @@ -20,39 +18,30 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; /// /// Central-side end of the central↔node command boundary (per-cluster mesh Phase 2). Routes -/// every central→node command out — over DistributedPubSub, ClusterClient or gRPC, per -/// — and forwards inbound s to the -/// deploy coordinator singleton. +/// every central→node command out — over DistributedPubSub (pair-local) or gRPC (cross-mesh), +/// per — and forwards inbound s to +/// the deploy coordinator singleton. /// /// /// -/// Registered per node, NOT as a cluster singleton. A driver node's ClusterClient -/// rotates across contact points and must find a live comm actor at whichever central node -/// answers. A singleton would be reachable only through the one node hosting it, and the -/// rotation would fail silently against the other. +/// Registered per node, NOT as a cluster singleton. A driver node posts its ack to +/// whichever central node answers first, so a live comm actor must exist on both. A singleton +/// would be reachable only through the one node hosting it, and the other central node's ack +/// ingress would have nothing to forward to. /// /// -/// 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. +/// The gRPC branch fans node by node over the contact load's target set. That load +/// selects exactly the enabled, non-maintenance ClusterNode rows carrying a +/// GrpcPort, and the fan dials every one of them, one call per node. It is deliberately +/// NOT narrower: the node side has no ClusterId or node filter at all — scoping happens +/// later, inside the artifact — so a node left out here silently keeps its old configuration +/// while the deployment seals green. /// /// -/// The gRPC branch fans over the same recipient set, node by node -/// (ClusterClient→gRPC Phase 1A). SendToAll reaches every node registered on a mesh's -/// receptionist, and the contact load already selects exactly the enabled, non-maintenance -/// rows those nodes come from; the gRPC branch dials that same row set directly, one call per -/// node, so the recipient set is identical by construction rather than by coincidence. It is -/// deliberately NOT narrower: the node side has no ClusterId or node filter at all — -/// scoping happens later, inside the artifact — so a node left out here silently keeps its -/// old configuration while the deployment seals green. -/// -/// -/// Fire-and-forget, on every transport. The gRPC branch pipes per-node results back to +/// Fire-and-forget, on both transports. The gRPC branch pipes per-node results back to /// itself for logging ONLY: a failure names the node and the status and is dropped. No retry, /// no queue, no ordering guarantee across nodes. That is not an omission — it is the -/// ClusterClient semantic preserved exactly, and the backstops are unchanged (the deploy +/// DistributedPubSub semantic preserved exactly, and the backstops are unchanged (the deploy /// coordinator's apply deadline, and a node's boot-time replay). /// /// @@ -62,7 +51,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers private readonly IDbContextFactory _dbFactory; private readonly MeshTransportOptions _options; - private readonly IMeshClusterClientFactory _clientFactory; private readonly Func _coordinator; // Resolved in PreStart, never at construction. See the Props doc: the DI graph behind these two @@ -88,15 +76,8 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers _dialer ??= Resolve(_dialerFactory, nameof(IMeshCommandDialer), firstUse: true); private readonly ILoggingAdapter _log = Context.GetLogger(); - private readonly bool _useClusterClient; private readonly bool _useGrpc; - // 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(); - // Node ids already warned about for a null GrpcPort. A persistently portless row would otherwise // produce one warning per refresh forever; same warn-once shape as TelemetryNodeSource. Actor // thread only. @@ -121,7 +102,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// Creates the props for this actor. /// Factory for the config database holding ClusterNode rows. /// The bound mesh-transport options. - /// Creates the ClusterClient; substituted in tests. /// /// Resolves the deploy-coordinator singleton proxy. A rather than /// an so this actor never depends on registration order at wiring @@ -129,9 +109,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// /// /// Resolves the thing that places one command call on one node, under - /// . Optional so the two older transports need - /// nothing new wired; a dialer under Grpc is a mis-wiring and - /// is reported loudly rather than silently dropping every command. + /// . Optional so the Dps transport needs nothing + /// new wired; a dialer under Grpc is a mis-wiring and is + /// reported loudly rather than silently dropping every command. /// /// /// Resolves central's per-cluster key lookup, told on every contact refresh which clusters @@ -163,37 +143,30 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers public static Props Props( IDbContextFactory dbFactory, MeshTransportOptions options, - IMeshClusterClientFactory clientFactory, Func coordinator, Func? dialer = null, Func? pskProvider = null) => Akka.Actor.Props.Create(() => - new CentralCommunicationActor( - dbFactory, options, clientFactory, coordinator, dialer, pskProvider)); + new CentralCommunicationActor(dbFactory, options, coordinator, dialer, pskProvider)); /// Initializes a new instance of the class. /// Factory for the config database. /// The bound mesh-transport options. - /// Creates the ClusterClient. /// Lazily resolves the deploy-coordinator singleton proxy. /// Resolves the gRPC command dialer; invoked in . /// Resolves the PSK provider; invoked in . public CentralCommunicationActor( IDbContextFactory dbFactory, MeshTransportOptions options, - IMeshClusterClientFactory clientFactory, Func coordinator, Func? dialer = null, Func? pskProvider = null) { _dbFactory = dbFactory; _options = options; - _clientFactory = clientFactory; _coordinator = coordinator; _dialerFactory = dialer; _pskProviderFactory = pskProvider; - _useClusterClient = string.Equals( - options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase); _useGrpc = string.Equals( options.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase); @@ -246,13 +219,13 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers _dialer = Resolve(_dialerFactory, nameof(IMeshCommandDialer), firstUse: false); _pskProvider = Resolve(_pskProviderFactory, nameof(IMeshPskProvider), firstUse: false); - if (!_useClusterClient && !_useGrpc) + if (!_useGrpc) { _log.Info( - "Mesh transport is {Mode}: commands are published on DistributedPubSub and no " - + "ClusterClient is created. The contact refresh still runs — it is what keeps the " - + "known-cluster gate current, and that gate belongs to the inbound ack endpoint, " - + "which is mapped by port and does not care about this mode", _options.Mode); + "Mesh transport is {Mode}: commands are published on DistributedPubSub and nothing " + + "is dialled. The contact refresh still runs — it is what keeps the known-cluster " + + "gate current, and that gate belongs to the inbound ack endpoint, which is mapped " + + "by port and does not care about this mode", _options.Mode); } // Deliberately NO null guard for either dependency here. At this point a still-null value @@ -266,7 +239,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers // Started in EVERY mode. The refresh is not transport state: it is central's ClusterNode // inventory, and the known-cluster gate it feeds protects an endpoint mapped by PORT, not by // MeshTransport:Mode. Gating the timer on the mode would mean a dark-switch rollback to - // ClusterClient or Dps left that endpoint live with the gate permanently inactive — silently + // Dps left that endpoint live with the gate permanently inactive — silently // reopening both the timing oracle and the pre-auth store/audit flood. Under Dps this is a // ConfigDb query every ContactRefreshSeconds that did not run before; central always has a // ConfigDb, and one indexed read a minute is the cheaper half of the trade. @@ -357,9 +330,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// protected override void PostStop() { - foreach (var (client, _) in _clients.Values) Context.Stop(client); - _clients.Clear(); - // Cancelling the lifecycle token also cancels every gRPC dial in flight. The dialer itself is // deliberately NOT disposed here: it is a container-owned singleton shared with anything else // that may dial, and an actor restart must not tear its channel pool down. @@ -369,43 +339,15 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers private void HandleMeshCommand(MeshCommand cmd) { - // Checked before the DPS arm: _useClusterClient is false under Grpc too, so the order is - // what keeps a Grpc-mode command off DistributedPubSub. + // Checked first, so a Grpc-mode command never reaches DistributedPubSub. DPS is mesh-wide and + // cannot cross an application-Cluster boundary at all in the split topology. if (_useGrpc) { DispatchOverGrpc(cmd); return; } - if (!_useClusterClient) - { - DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, cmd.Message)); - return; - } - - 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 - // apply deadline naming the silent nodes, which is a better operator signal than a - // 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 in any " - + "cluster; the command is not buffered and will not be retried", - cmd.Message.GetType().Name); - return; - } - - // 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)); + DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, cmd.Message)); } /// @@ -453,10 +395,10 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers var targets = _targets; if (targets.Count == 0) { - // App-level drop, exactly as the ClusterClient arm does with no client. The command is - // NOT queued and will NOT be retried: a deploy fails at the coordinator's apply deadline - // naming the silent nodes, which is a better operator signal than a command arriving - // minutes late against a config the DB already recorded as failed. + // App-level drop. The command is NOT queued and will NOT be retried: a deploy fails at + // the coordinator's apply deadline naming the silent nodes, which is a better operator + // signal than a command arriving minutes late against a config the DB already recorded + // as failed. _log.Warning( "No mesh gRPC targets — dropping {MessageType}. The last contact refresh produced no " + "enabled, non-maintenance ClusterNode row carrying a GrpcPort in any cluster; the " @@ -632,7 +574,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers return; // Stopping. } - var systemName = Context.System.Name; var dbFactory = _dbFactory; // Off the actor thread, result piped back as a message. A synchronous DB read here would @@ -641,29 +582,21 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers { await using var db = await dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false); // UNFILTERED, with the eligibility flags projected and applied per row below. The two - // outputs of this read want different row sets: the contact/dispatch sets are the - // enabled, non-maintenance nodes central talks TO, while the known-cluster gate answers - // "is this a cluster central has ever heard of" and must include a cluster whose every - // node is disabled or in maintenance — such a cluster can still send an ack, and denying - // it would make maintenance mode an authentication switch. + // outputs of this read want different row sets: the dispatch set is the enabled, + // non-maintenance nodes central talks TO, while the known-cluster gate answers "is this a + // cluster central has ever heard of" and must include a cluster whose every node is + // disabled or in maintenance — such a cluster can still send an ack, and denying it would + // make maintenance mode an authentication switch. var rows = await db.ClusterNodes .AsNoTracking() .Select(n => new { - n.NodeId, n.ClusterId, n.Host, n.AkkaPort, n.GrpcPort, n.Enabled, n.MaintenanceMode, + n.NodeId, n.ClusterId, n.Host, n.GrpcPort, n.Enabled, n.MaintenanceMode, }) .ToListAsync(ct) .ConfigureAwait(false); - // 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(); - - // The gRPC dispatch set, built from the SAME row set in the SAME pass. Sharing the read is - // what makes "the gRPC fan reaches exactly the ClusterClient recipient set" structural - // rather than a claim two separate queries would have to keep agreeing on. + // The gRPC dispatch set: one entry per eligible row that carries a port. var targets = new List(rows.Count); var portless = new List(); @@ -677,8 +610,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers clusterIds.Add(row.ClusterId); // The eligibility filter that used to be the query's WHERE clause. Everything below - // this line — receptionist contacts and gRPC targets — sees exactly the rows it saw - // before. + // this line — the gRPC targets — sees exactly the rows it saw before. if (!row.Enabled || row.MaintenanceMode) continue; @@ -687,29 +619,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers row.NodeId, row.ClusterId, $"http://{row.Host}:{grpcPort}")); else portless.Add(row.NodeId); - - var address = $"akka.tcp://{systemName}@{row.Host}:{row.AkkaPort}/system/receptionist"; - - // Parse up front, per row, inside the loop's own guard. A single malformed row must - // 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 _)) - { - 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}"); - } } - var contactsByCluster = byCluster.ToDictionary( - kv => kv.Key, kv => (IReadOnlyList)kv.Value); - return new ContactsLoaded( - contactsByCluster, malformed, targets, portless, clusterIds.ToList()); + return new ContactsLoaded(targets, portless, clusterIds.ToList()); }, ct).PipeTo(self); } @@ -722,66 +634,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers // EVERY mode. See PreStart — the gate protects an endpoint mapped by port, not by mode. PublishKnownClusters(msg); - - if (!_useClusterClient) - return; // Dps or Grpc: no ClusterClient to rebuild. - - foreach (var bad in msg.Malformed) - { - _log.Warning( - "ClusterNode {Row} does not yield a usable receptionist address; skipping it in this " - + "refresh (other nodes are unaffected). Check the row's Host and AkkaPort", bad); - } - - var byCluster = msg.ContactsByCluster.ToDictionary( - kv => kv.Key, kv => kv.Value.ToImmutableHashSet()); - - // 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. LoadContactsFromDb only - // creates a ByCluster key when at least one of its rows parsed, so an all-malformed cluster has - // NO key at all and is caught here by the absent-key (!TryGetValue) arm — not by an empty value. - var gone = _clients.Keys - .Where(id => !byCluster.TryGetValue(id, out var c) || c.IsEmpty) - .ToList(); - foreach (var clusterId in gone) - { - _log.Info( - "Cluster {ClusterId} no longer yields any usable contact point; stopping its " - + "ClusterClient", clusterId); - Context.Stop(_clients[clusterId].Client); - _clients.Remove(clusterId); - } - - var anyUsable = false; - foreach (var (clusterId, contacts) in byCluster) - { - if (contacts.IsEmpty) - { - // Defensive: LoadContactsFromDb never emits an empty-but-present cluster (a key is - // created only on a first successfully-parsed row), so this arm is unreachable for the - // DB producer — an all-malformed cluster is handled by the absent-key branch above. It - // guards a ContactsLoaded (a public record) constructed elsewhere with an empty list. - _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; - } - - 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"); - } } /// @@ -790,11 +642,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// /// The load result. /// - /// The malformed-receptionist-address warnings are deliberately NOT logged here: nothing on - /// this transport dials a receptionist, and reporting an unusable Akka address as a - /// command-plane problem would send an operator after the wrong column. A host that is - /// unusable as a gRPC endpoint surfaces instead as a named per-node dial failure on every - /// dispatch, which is a truer signal. + /// A host that is unusable as a gRPC endpoint is deliberately NOT screened here: it surfaces + /// as a named per-node dial failure on every dispatch, which is a truer signal than a + /// refresh-time warning about a string that may well be fine. /// private void AdoptGrpcTargets(ContactsLoaded msg) { @@ -837,7 +687,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// Not gated on , on purpose. The gate feeds /// CentralCommandAuthInterceptor, which is registered on every admin node, and the /// ack service it protects is mapped by listener port — neither consults the mode. Gating - /// this on Grpc would mean a dark-switch rollback to ClusterClient or + /// this on Grpc would mean a dark-switch rollback to /// Dps left that endpoint serving with a permanently inactive gate: the timing /// oracle and the pre-authentication store/audit flood would both come back, silently, /// as a side effect of a rollback nobody would connect to authentication. @@ -887,75 +737,13 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers pskProvider.InvalidateAll(); } - /// Stops a cluster's existing client and creates a replacement for its new contact set. - /// - /// - /// 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. - /// - /// - /// 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 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 application Cluster whose client is being (re)built. - /// The receptionist addresses to dial for that cluster. - private void RebuildClient(string clusterId, ImmutableHashSet contacts) - { - if (_clients.TryGetValue(clusterId, out var existing)) - { - _log.Info("Contact set for cluster {ClusterId} changed; replacing its ClusterClient", clusterId); - Context.Stop(existing.Client); - - // Removed now: if the create below throws, a stale ref would route commands into a - // stopping actor and they would vanish without a warning. - _clients.Remove(clusterId); - } - - try - { - var paths = contacts.Select(ActorPath.Parse).ToImmutableHashSet(); - 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 for cluster {ClusterId}; its nodes' commands " - + "are dropped until the next contact refresh succeeds", clusterId); - } - } - /// Internal tick asking for a contact-point refresh. public sealed record RefreshContacts; /// Result of a contact-point load, piped back from the DB read. - /// - /// 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. /// /// The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a - /// GrpcPort. Built from the same rows as . + /// GrpcPort. /// /// /// Node ids skipped because their row has no GrpcPort. Carried rather than logged in @@ -968,8 +756,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers /// still send an ack, and refusing it would make maintenance mode an authentication switch. /// public sealed record ContactsLoaded( - IReadOnlyDictionary> ContactsByCluster, - IReadOnlyList Malformed, IReadOnlyList Targets, IReadOnlyList PortlessNodes, IReadOnlyList ClusterIds); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs deleted file mode 100644 index 83b2b2ee..00000000 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs +++ /dev/null @@ -1,67 +0,0 @@ -using System.Collections.Immutable; -using Akka.Actor; -using Akka.Cluster.Tools.Client; - -namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; - -/// -/// Creates the ClusterClient central uses to reach driver nodes. Exists as a seam so -/// CentralCommunicationActor can be tested against a TestProbe without standing up -/// a second cluster. -/// -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, string clusterId, ImmutableHashSet contacts); -} - -/// -public sealed class DefaultMeshClusterClientFactory : IMeshClusterClientFactory -{ - /// - /// Per-incarnation generation counter, appended to the actor name. - /// - /// - /// Context.Stop of the previous client is asynchronous — its actor name stays - /// reserved until termination completes — so recreating the same name while handling a single - /// message throws . That is exactly what a contact-set - /// change does: stop the old client, create the replacement, both in one handler. A - /// generation suffix makes every incarnation's name unique by construction. Ported from the - /// sister project, where it was a shipped bug fix rather than foresight. - /// - private long _generation; - - /// - 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-{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/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 1801da38..472f753e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -56,11 +56,11 @@ public static class ServiceCollectionExtensions // 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. + // A plain WithActors, NOT a singleton — and that is the load-bearing part. A driver node + // walks MeshTransport:CentralGrpcEndpoints in order and posts its ack to whichever central + // node answers first, so a live comm actor must exist on both. As a singleton it would exist + // on one central node only, and the other node's ack ingress would have nothing to forward + // to: acks would land sometimes, depending on which endpoint answered. builder.WithActors((system, registry, resolver) => { var dbFactory = resolver.GetService>(); @@ -85,7 +85,6 @@ public static class ServiceCollectionExtensions 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(out var coordinator) 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 4a0bdc57..c93cabd8 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 @@ -1,9 +1,6 @@ -using System.Collections.Immutable; using System.Diagnostics.CodeAnalysis; using Akka.Actor; -using Akka.Cluster.Tools.Client; using Akka.Cluster.Tools.PublishSubscribe; -using Akka.TestKit; using Microsoft.EntityFrameworkCore; using Shouldly; using Xunit; @@ -25,40 +22,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase { private const string DeploymentsTopic = "deployments"; - /// Records every client the actor asks for, and hands back a probe in its place. - private sealed class RecordingClientFactory : IMeshClusterClientFactory - { - private readonly Func _newProbe; - - public RecordingClientFactory(Func newProbe) => _newProbe = newProbe; - - public List> Calls { get; } = []; - - public List ClusterIds { get; } = []; - - public List Probes { get; } = []; - - /// 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; - } - } - - private static MeshTransportOptions ClusterClientMode(int refreshSeconds = 60) => new() - { - Mode = MeshTransportOptions.ModeClusterClient, - CentralContactPoints = ["akka.tcp://otopcua@central-1:4053"], - ContactRefreshSeconds = refreshSeconds, - }; - private static MeshTransportOptions GrpcMode(int refreshSeconds = 60) => new() { Mode = MeshTransportOptions.ModeGrpc, @@ -66,47 +29,7 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase DispatchTimeoutSeconds = 10, }; - private static void SeedCluster( - IDbContextFactory factory, - string clusterId, - params (string NodeId, string Host, int AkkaPort, bool Enabled, bool Maintenance)[] nodes) - { - using var db = factory.CreateDbContext(); - db.ServerClusters.Add(new ServerCluster - { - ClusterId = clusterId, - Name = clusterId, - Enterprise = "ZB", - Site = "Site", - // Warm/Hot require NodeCount = 2 (CK_ServerCluster_RedundancyMode_NodeCount); these - // fixtures only need the row to exist, so use the standalone shape. - RedundancyMode = RedundancyMode.None, - NodeCount = 1, - CreatedBy = "test", - }); - foreach (var n in nodes) - { - db.ClusterNodes.Add(new ClusterNode - { - NodeId = n.NodeId, - ClusterId = clusterId, - Host = n.Host, - AkkaPort = n.AkkaPort, - ApplicationUri = $"urn:{n.NodeId}", - Enabled = n.Enabled, - MaintenanceMode = n.Maintenance, - CreatedBy = "test", - }); - } - - db.SaveChanges(); - } - - /// - /// Seeds a cluster whose nodes carry a GrpcPort — the gRPC command plane's shape. - /// Kept separate from so the ClusterClient facts, which have no - /// opinion about the port, stay exactly as they were. - /// + /// Seeds a cluster whose nodes carry a GrpcPort — the command plane's shape. private static void SeedGrpcCluster( IDbContextFactory factory, string clusterId, @@ -145,15 +68,16 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase private IActorRef Spawn( IDbContextFactory factory, MeshTransportOptions options, - IMeshClusterClientFactory clientFactory, IActorRef? coordinator = null) => - Sys.ActorOf(CentralCommunicationActor.Props(factory, options, clientFactory, () => coordinator)); + Sys.ActorOf(CentralCommunicationActor.Props(factory, options, () => coordinator)); [Fact] - public void Dps_mode_publishes_on_the_carried_topic_and_creates_no_client() + public void Dps_mode_publishes_on_the_carried_topic() { + // Dps is the pair-local default and stays in the tree: it is the correct transport for a + // single-mesh topology, which every integration test and every profile that leaves + // MeshTransport:Mode unset still runs. var factory = NewInMemoryDbFactory(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); var mediatorProbe = CreateTestProbe(); // Subscribe on the probe's behalf: SubscribeAck goes to the SENDER, so telling the mediator // without an explicit sender would ack to TestActor and this wait would hang. @@ -161,150 +85,22 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref); mediatorProbe.ExpectMsg(TimeSpan.FromSeconds(5)); - var actor = Spawn(factory, new MeshTransportOptions(), clients); + var actor = Spawn(factory, new MeshTransportOptions()); var dispatch = new DispatchDeployment( new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid())); actor.Tell(new MeshCommand(DeploymentsTopic, dispatch)); mediatorProbe.ExpectMsg(TimeSpan.FromSeconds(5)).ShouldBe(dispatch); - clients.Calls.ShouldBeEmpty(); - } - - [Fact] - public void ClusterClient_mode_sends_to_all_not_to_one() - { - // THE assertion of this phase. ClusterClient.Send delivers to exactly ONE registered actor; - // today's DPS publish reaches EVERY DriverHostActor, and the node side has no ClusterId or - // node filter to compensate. Send would deploy to a single node while every other node - // silently kept its old configuration. - var factory = NewInMemoryDbFactory(); - SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false)); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - - var actor = Spawn(factory, ClusterClientMode(), clients); - AwaitCondition(() => clients.Probes.Count == 1, TimeSpan.FromSeconds(5)); - - var dispatch = new DispatchDeployment( - new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid())); - actor.Tell(new MeshCommand(DeploymentsTopic, dispatch)); - - var sent = clients.Probes[0].ExpectMsg(TimeSpan.FromSeconds(5)); - sent.Path.ShouldBe(MeshPaths.NodeCommunication); - sent.Message.ShouldBe(dispatch); - } - - [Fact] - public void One_client_per_cluster_is_created_for_a_multi_cluster_fleet() - { - // 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 == 2, TimeSpan.FromSeconds(5)); - - // Give a second refresh a chance to (wrongly) create more. - ExpectNoMsg(TimeSpan.FromMilliseconds(300)); - 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] - public void Disabled_and_maintenance_nodes_are_not_dialled() - { - var factory = NewInMemoryDbFactory(); - SeedCluster( - factory, - "C1", - ("host-a:4053", "host-a", 4053, true, false), - ("host-b:4053", "host-b", 4053, false, false), - ("host-c:4053", "host-c", 4053, true, true)); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - - Spawn(factory, ClusterClientMode(), clients); - AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5)); - - clients.Calls[0].Count.ShouldBe(1); - clients.Calls[0].Single().ToString().ShouldContain("host-a"); - } - - [Fact] - public void A_malformed_node_row_does_not_abort_the_refresh() - { - // The bad row is ordered FIRST on purpose: the failure this guards against is an exception - // that aborts the loop and leaves the contact set half-built. - var factory = NewInMemoryDbFactory(); - SeedCluster( - factory, - "C1", - ("bad", "not a host!!", 4053, true, false), - ("host-a:4053", "host-a", 4053, true, false)); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - - Spawn(factory, ClusterClientMode(), clients); - AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5)); - - clients.Calls[0].ShouldContain(p => p.ToString().Contains("host-a")); - } - - [Fact] - public void A_command_with_no_client_is_dropped_with_a_warning() - { - // No ClusterNode rows at all: nothing to dial, so the command must be dropped loudly rather - // than queued. buffer-size = 0 covers the case where a client exists but cannot connect; - // this covers the case where there is no client at all. - var factory = NewInMemoryDbFactory(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - var actor = Spawn(factory, ClusterClientMode(), clients); - - EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () => - actor.Tell(new MeshCommand(DeploymentsTopic, "anything"))); } [Fact] public void ApplyAck_is_forwarded_to_the_coordinator_preserving_the_sender() { var factory = NewInMemoryDbFactory(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); var coordinator = CreateTestProbe(); var node = CreateTestProbe(); - var actor = Spawn(factory, ClusterClientMode(), clients, coordinator.Ref); + var actor = Spawn(factory, GrpcMode(), coordinator.Ref); var ack = new ApplyAck( new DeploymentId(Guid.NewGuid()), @@ -324,8 +120,7 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase public void An_unresolvable_coordinator_logs_rather_than_losing_the_ack_silently() { var factory = NewInMemoryDbFactory(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - var actor = Spawn(factory, ClusterClientMode(), clients, coordinator: null); + var actor = Spawn(factory, GrpcMode(), coordinator: null); var ack = new ApplyAck( new DeploymentId(Guid.NewGuid()), @@ -342,12 +137,11 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase public void A_faulted_contact_load_warns_and_leaves_the_actor_alive() { var factory = NewInMemoryDbFactory(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); - var actor = Spawn(factory, ClusterClientMode(), clients); + var actor = Spawn(factory, GrpcMode()); // Filter on a phrase unique to the DB-failure warning. EventFilter matches case-INSENSITIVELY, - // so a looser "was NOT" also caught this actor's other warning ("the mesh ClusterClient was - // not created") and the assertion failed with two matches instead of one. + // so a looser "was NOT" would also catch this actor's other warnings and the assertion would + // fail with two matches instead of one. EventFilter.Warning(contains: "contact set was NOT refreshed").ExpectOne( TimeSpan.FromSeconds(5), () => actor.Tell(new Status.Failure(new InvalidOperationException("db down")))); @@ -359,16 +153,15 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase // ------------------------------------------------------------------ Mode = Grpc // - // The third transport. The fan-out must reach EXACTLY the recipient set SendToAll reaches today - // — every enabled, non-maintenance node in every cluster — because the node side still has no - // ClusterId or node filter; scoping happens later, inside the artifact. A narrower fan would - // leave nodes silently on their old configuration while the deployment sealed green. + // The cross-mesh transport. The fan-out must reach EVERY enabled, non-maintenance node in every + // cluster, because the node side has no ClusterId or node filter at all; scoping happens later, + // inside the artifact. A narrower fan would leave nodes silently on their old configuration + // while the deployment sealed green. [Fact] public void Grpc_mode_fans_one_command_to_every_enabled_node_across_all_clusters() { - // THE assertion of this task, and the direct counterpart of - // A_command_fans_out_a_send_to_all_to_every_cluster_client. + // THE assertion of this transport: one command, one call per node, across every cluster. var factory = NewInMemoryDbFactory(); SeedGrpcCluster( factory, @@ -392,31 +185,32 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase } [Fact] - public void Grpc_mode_creates_no_ClusterClient_and_publishes_nothing_on_pubsub() + public void Grpc_mode_publishes_nothing_on_pubsub() { + // DPS is mesh-wide and cannot cross an application-Cluster boundary, so a command that also + // went out on the topic would be delivered twice inside central's own mesh and not at all + // anywhere else. var factory = NewInMemoryDbFactory(); SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); - var clients = new RecordingClientFactory(() => CreateTestProbe()); var mediatorProbe = CreateTestProbe(); DistributedPubSub.Get(Sys).Mediator.Tell( new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref); mediatorProbe.ExpectMsg(TimeSpan.FromSeconds(5)); - var rig = SpawnGrpcRig(factory, clientFactory: clients); + var rig = SpawnGrpcRig(factory); AwaitContactsLoaded(rig); rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())); AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5)); - clients.Calls.ShouldBeEmpty(); mediatorProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } [Fact] public void Grpc_mode_does_not_dial_disabled_or_maintenance_nodes() { - // Same recipient filter as the ClusterClient contact load, asserted independently so the two - // transports cannot drift on who counts as a node. + // Disabled and maintenance-mode rows are not nodes central talks to; the same filter the + // contact load applies, asserted here so the fan cannot drift from it. var factory = NewInMemoryDbFactory(); SeedGrpcCluster( factory, @@ -725,22 +519,20 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase [InlineData(MeshTransportOptions.ModeGrpc)] public void The_known_cluster_gate_is_published_in_every_transport_mode(string mode) { - // Replaces an earlier fact asserting the OPPOSITE, whose premise did not survive 1B: the + // Replaces an earlier fact asserting the OPPOSITE, whose premise did not survive Phase 1B: the // gate feeds CentralCommandAuthInterceptor, which is registered on every admin node, and the // ack service it protects is mapped by LISTENER PORT — neither consults MeshTransport:Mode. - // Gating the publication on Grpc meant a dark-switch rollback to ClusterClient or Dps left - // that endpoint serving with a permanently inactive gate, silently restoring both the timing + // Gating the publication on Grpc meant a dark-switch rollback to Dps left that endpoint + // serving with a permanently inactive gate, silently restoring both the timing // oracle and the pre-authentication store/audit flood. The set is central's cluster // inventory, not transport state. var factory = NewInMemoryDbFactory(); SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false)); var psk = new RecordingPskProvider(); - var clients = new RecordingClientFactory(() => CreateTestProbe()); Sys.ActorOf(CentralCommunicationActor.Props( factory, new MeshTransportOptions { Mode = mode, ContactRefreshSeconds = 60 }, - clients, () => null, () => new RecordingDialer(), () => psk)); @@ -749,12 +541,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase () => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5)); psk.KnownClusterPublications[0].ShouldBe(new[] { "C1" }); psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1); - - // The one load-bearing line the publication itself cannot witness: with the refresh - // timer now running in every mode, only the ClusterClient branch may construct clients. - // With rows seeded, deleting the !_useClusterClient guard turns this red for Dps/Grpc. - if (mode != MeshTransportOptions.ModeClusterClient) - clients.Calls.ShouldBeEmpty(); } [Fact] @@ -815,7 +601,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase var actor = Sys.ActorOf(CentralCommunicationActor.Props( factory, GrpcMode(), - new RecordingClientFactory(() => CreateTestProbe()), () => null, dialer: null, pskProvider: null)); @@ -871,7 +656,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase actor = Sys.ActorOf(CentralCommunicationActor.Props( factory, GrpcMode(), - new RecordingClientFactory(() => CreateTestProbe()), () => null, DialerFactory, PskFactory)); @@ -924,7 +708,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase actor = Sys.ActorOf(CentralCommunicationActor.Props( factory, GrpcMode(), - new RecordingClientFactory(() => CreateTestProbe()), () => null, () => new RecordingDialer(), pskProvider: PskFactory)); @@ -969,7 +752,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase actor = Sys.ActorOf(CentralCommunicationActor.Props( factory, GrpcMode(), - new RecordingClientFactory(() => CreateTestProbe()), () => null, dialer: () => throw new InvalidOperationException("permanently broken"), pskProvider: () => psk)); @@ -1006,7 +788,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase private GrpcRig SpawnGrpcRig( IDbContextFactory factory, RecordingDialer? dialer = null, - IMeshClusterClientFactory? clientFactory = null, IActorRef? coordinator = null) { dialer ??= new RecordingDialer(); @@ -1014,7 +795,6 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase var actor = Sys.ActorOf(CentralCommunicationActor.Props( factory, GrpcMode(), - clientFactory ?? new RecordingClientFactory(() => CreateTestProbe()), () => coordinator, () => dialer, () => psk)); 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 deleted file mode 100644 index 3c188719..00000000 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs +++ /dev/null @@ -1,82 +0,0 @@ -using System.Collections.Immutable; -using Akka.Actor; -using Akka.Cluster.Tools.Client; -using Akka.Configuration; -using Akka.TestKit.Xunit2; -using Shouldly; -using Xunit; -using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; - -namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; - -/// -/// Guards the one non-obvious part of client creation: recreating a client with the same name -/// inside a single message handler throws, because Context.Stop is asynchronous and the -/// name stays reserved until termination completes. -/// -public class MeshClusterClientFactoryTests : TestKit -{ - private static readonly Config TestConfig = ConfigurationFactory.ParseString(@" - akka.actor.provider = cluster - akka.remote.dot-netty.tcp.port = 0 - akka.remote.dot-netty.tcp.hostname = 127.0.0.1") - .WithFallback(ClusterClientReceptionist.DefaultConfig()); - - public MeshClusterClientFactoryTests() - : base(TestConfig) - { - } - - // A real ClusterClient pointed at an unreachable contact is safe in TestKit — it simply retries - // the initial-contact rotation and never fails the test. - private static ImmutableHashSet UnreachableContact => ImmutableHashSet.Create( - ActorPath.Parse("akka.tcp://otopcua@127.0.0.1:65501/system/receptionist")); - - [Fact] - public void Recreating_a_client_in_one_pass_does_not_collide_on_the_actor_name() - { - var factory = new DefaultMeshClusterClientFactory(); - - 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, "C1", UnreachableContact); - - second.Path.Name.ShouldNotBe(first.Path.Name); - } - - [Fact] - public void Every_incarnation_gets_a_distinct_name() - { - var factory = new DefaultMeshClusterClientFactory(); - - var names = Enumerable.Range(0, 5) - .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 - // real two-ActorSystem boundary test, which only delivers if the contacts reached the client. -} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs deleted file mode 100644 index 267dd3ce..00000000 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs +++ /dev/null @@ -1,446 +0,0 @@ -using System.Collections.Immutable; -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; - -namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; - -/// -/// Drives a real ClusterClient across two genuinely separate s — -/// the only test in the repo where the mesh boundary carries a message over the wire. -/// -/// -/// -/// The sister project substitutes an in-process relay that unwraps -/// and forwards. That tests actor logic and cannot -/// fail for any transport reason: not a missing receptionist extension, not a service that -/// was never registered, not a malformed contact path, not a serialization break. Those are -/// precisely the Phase 2 failure modes, so they are what this test exercises. -/// -/// -/// Two clusters, one ActorSystem name, joined ONLY by ClusterClient. The shared name -/// is mandatory — Akka.Remote matches on the full address, so a contact path naming -/// otopcua cannot reach a system called anything else. Each system seeds itself and -/// neither lists the other, and -/// asserts the -/// separation outright: if they ever gossiped into one mesh, every delivery below could be -/// plain local messaging and the whole class would be testing nothing. -/// -/// -/// Scope limit. With exactly one node per side, "registered per node" and "registered -/// as a cluster singleton" are indistinguishable by construction, so this class does NOT -/// cover the per-node property whose structural assertions MeshCommActorPathTests -/// dropped. That property's discriminating coverage is live-gate step 8: stop the OLDEST -/// central and the survivor's comm actor must still receive a node ack, which a singleton -/// hosted on the stopped node cannot. -/// -/// -public sealed class MeshClusterClientBoundaryTests : IAsyncLifetime -{ - private const string SharedSystemName = "otopcua"; - private static readonly TimeSpan Arrival = TimeSpan.FromSeconds(30); - - /// - /// How long the falsifiability control waits before concluding nothing arrived. Deliberately - /// shorter than but far longer than a successful delivery takes in - /// practice, so a pass means "the boundary is shut", not "the clock ran out first". - /// - private static readonly TimeSpan Silence = TimeSpan.FromSeconds(10); - - /// - /// One factory for every client this class creates, because the generation counter that keeps - /// client actor names unique is per factory instance. A second factory on the same - /// system restarts at generation 1 and collides with the first factory's client — which is - /// exactly how this test first failed. Production has one instance per - /// CentralCommunicationActor, so the real code path is safe. - /// - private readonly DefaultMeshClusterClientFactory _clientFactory = new(); - - private ActorSystem _central = null!; - private ActorSystem _node = null!; - private IActorRef _nodeComm = null!; - private IActorRef _centralClient = null!; - private TaskCompletionSource _centralInbox = null!; - - /// - public async ValueTask InitializeAsync() - { - _central = StartSelfSeededSystem(withReceptionist: true); - _node = StartSelfSeededSystem(withReceptionist: true); - await Task.WhenAll(AwaitSingleMemberUpAsync(_central), AwaitSingleMemberUpAsync(_node)); - - // Central: the ack landing pad, at the path the node's comm actor sends to. - _centralInbox = NewInbox(); - var centralComm = _central.ActorOf( - CaptureActor.Props(_centralInbox), MeshPaths.CentralCommunicationName); - ClusterClientReceptionist.Get(_central).RegisterService(centralComm); - - // Node: the REAL comm actor, dialling central through the REAL production client factory. - 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, "site-a", ReceptionistOf(_node)); - } - - /// - public async ValueTask DisposeAsync() - { - await ShutdownAsync(_central); - await ShutdownAsync(_node); - } - - [Fact] - public void The_two_systems_are_separate_clusters_joined_only_by_the_client() - { - // The load-bearing precondition of every other test here. Two systems sharing a name in one - // process are one gossip accident away from being one cluster, at which point ClusterClient - // is no longer the thing under test. - AkkaCluster.Get(_central).State.Members.Count.ShouldBe(1, - "central must be its own single-member cluster, or the boundary is not being crossed"); - AkkaCluster.Get(_node).State.Members.Count.ShouldBe(1, - "the node must be its own single-member cluster, or the boundary is not being crossed"); - } - - [Fact] - public async Task Central_send_to_all_reaches_the_node_comm_actor_and_lands_on_its_event_stream() - { - var inbox = NewInbox(); - _node.EventStream.Subscribe(_node.ActorOf(CaptureActor.Props(inbox)), typeof(RestartDriver)); - - var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid()); - var delivered = await SendUntilDeliveredAsync( - _centralClient, - () => new ClusterClient.SendToAll(MeshPaths.NodeCommunication, command), - inbox, - Arrival); - - delivered.ShouldNotBeNull( - "SendToAll must reach the node's registered comm actor and be re-emitted node-locally; " - + "this is the whole central→node leg of the mesh transport"); - delivered.ShouldBeOfType().DriverInstanceId.ShouldBe("driver-1"); - } - - [Fact] - public async Task A_node_ack_reaches_central_through_the_node_comm_actors_own_client() - { - // Told directly rather than routed from central, so a failure here can only be the outbound - // leg. The inbound leg has its own test above. - var ack = new ApplyAck( - new DeploymentId(Guid.NewGuid()), - new NodeId("node-a"), - ApplyAckOutcome.Applied, - null, - CorrelationId.NewId()); - - var delivered = await SendUntilDeliveredAsync( - _nodeComm, () => ack, _centralInbox, Arrival); - - delivered.ShouldNotBeNull( - "the node's ApplyAck must reach central's comm actor, or every deployment times out at " - + "the apply deadline naming nodes that applied successfully"); - delivered.ShouldBeOfType().NodeId.Value.ShouldBe("node-a"); - } - - [Fact] - public async Task Falsifiability_control_a_node_without_the_receptionist_extension_receives_nothing() - { - // Without this, the two tests above cannot distinguish "the ClusterClient boundary works" - // from "something else in the process delivered the message". Same message, same client - // factory, same paths, same comm actor — and it must NOT arrive. - // - // HONEST SCOPE: this removes the receptionist extension AND, unavoidably, the RegisterService - // call — resolving ClusterClientReceptionist.Get to register would materialise the very - // extension whose absence is the control. So it falsifies "delivery happens without a - // receptionist boundary", not the extension line in isolation. Verified falsifiable: giving - // this node the extension and the registration turns the test red, which also proves the send - // below is live rather than silently misaddressed. - var deaf = StartSelfSeededSystem(withReceptionist: false); - try - { - await AwaitSingleMemberUpAsync(deaf); - - var inbox = NewInbox(); - deaf.EventStream.Subscribe(deaf.ActorOf(CaptureActor.Props(inbox)), typeof(RestartDriver)); - - // Deliberately NOT ClusterClientReceptionist.Get(deaf) — resolving the extension would - // materialise the very receptionist whose absence is the control. - var comm = deaf.ActorOf(NodeCommunicationActor.Props(null), MeshPaths.NodeCommunicationName); - comm.ShouldNotBeNull(); - - var client = _clientFactory.Create(_central, "deaf", ReceptionistOf(deaf)); - var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid()); - - var deadline = DateTime.UtcNow + Silence; - while (DateTime.UtcNow < deadline && !inbox.Task.IsCompleted) - { - client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, command)); - await Task.Delay(500, TestContext.Current.CancellationToken); - } - - inbox.Task.IsCompleted.ShouldBeFalse( - "delivery survived removing the receptionist extension, so the two positive tests " - + "above are not measuring the ClusterClient boundary at all"); - } - finally - { - await ShutdownAsync(deaf); - } - } - - [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. - /// - /// - /// drops the receptionist provider from akka.extensions, - /// leaving DistributedPubSub in place so only the one variable changes. - /// - /// The started system. - private static ActorSystem StartSelfSeededSystem(bool withReceptionist) - { - var extensions = withReceptionist - ? string.Empty - : "akka.extensions = [\"Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, " - + "Akka.Cluster.Tools\"]\n"; - - // Port 0 + a post-start self-Join, rather than a seed-nodes list: the address is not known - // until the transport binds, and hard-coding ports makes the test collide with itself under - // parallel execution. - var config = ConfigurationFactory.ParseString( - "akka.remote.dot-netty.tcp.hostname = \"127.0.0.1\"\n" - + "akka.remote.dot-netty.tcp.public-hostname = \"127.0.0.1\"\n" - + "akka.remote.dot-netty.tcp.port = 0\n" - + "akka.cluster.seed-nodes = []\n" - + "akka.loglevel = \"WARNING\"\n" - + extensions) - .WithFallback(ZB.MOM.WW.OtOpcUa.Cluster.HoconLoader.LoadBaseConfig()); - - var system = ActorSystem.Create(SharedSystemName, config); - var cluster = AkkaCluster.Get(system); - cluster.Join(cluster.SelfAddress); - return system; - } - - private static async Task AwaitSingleMemberUpAsync(ActorSystem system) - { - var cluster = AkkaCluster.Get(system); - var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); - while (DateTime.UtcNow < deadline) - { - if (cluster.State.Members.Any(m => m.Status == MemberStatus.Up)) return; - await Task.Delay(100, TestContext.Current.CancellationToken); - } - - cluster.State.Members.Any(m => m.Status == MemberStatus.Up).ShouldBeTrue( - $"{cluster.SelfAddress} never reached Up as its own seed"); - } - - /// - /// Re-sends until the inbox completes or the deadline passes. - /// - /// - /// A ClusterClient establishes contact asynchronously and buffer-size = 0 means anything - /// sent before that lands is dropped on the floor — by design. A single send would therefore - /// race the handshake and fail intermittently for a reason that is not a defect. Re-sending - /// weakens nothing: the control test re-sends on exactly the same schedule and must still - /// receive nothing. - /// - /// Actor to send to. - /// Builds the message for each attempt. - /// Completion source the receiving capture actor resolves. - /// How long to keep trying. - /// The delivered message, or if none arrived. - private static async Task SendUntilDeliveredAsync( - IActorRef target, - Func message, - TaskCompletionSource inbox, - TimeSpan timeout) - { - var deadline = DateTime.UtcNow + timeout; - while (DateTime.UtcNow < deadline) - { - if (inbox.Task.IsCompleted) return await inbox.Task; - target.Tell(message()); - await Task.Delay(250, TestContext.Current.CancellationToken); - } - - return inbox.Task.IsCompleted ? await inbox.Task : null; - } - - private static TaskCompletionSource NewInbox() => - new(TaskCreationOptions.RunContinuationsAsynchronously); - - private static ImmutableHashSet ReceptionistOf(ActorSystem system) => - ImmutableHashSet.Create( - new RootActorPath(AkkaCluster.Get(system).SelfAddress) / "system" / "receptionist"); - - private static async Task ShutdownAsync(ActorSystem? system) - { - if (system is null) return; - await system.Terminate().WaitAsync(TimeSpan.FromSeconds(15)); - } - - /// Completes an inbox with the first message it receives. - private sealed class CaptureActor : ReceiveActor - { - public static Props Props(TaskCompletionSource sink) => - Akka.Actor.Props.Create(() => new CaptureActor(sink)); - - public CaptureActor(TaskCompletionSource sink) => ReceiveAny(m => sink.TrySetResult(m)); - } -}