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

refactor(mesh): step 1 — delete the central ClusterClient factory and per-cluster client cache

Browse Source 
IMeshClusterClientFactory/DefaultMeshClusterClientFactory are gone, along with
CentralCommunicationActor's _clients map, RebuildClient, the SendToAll fan and the
receptionist-address half of the contact load. LoadContactsFromDb SURVIVES — it feeds the
gRPC dialer's endpoint resolution and the known-cluster gate — but now projects only the
columns those two need (the AkkaPort/ActorPath.TryParse parse and its Malformed list went
with the receptionist addresses).

Ripples fixed in this step because the build must be green at every commit:
  - ControlPlane/ServiceCollectionExtensions drops the factory argument.
  - CentralCommunicationActorTests: the six ClusterClient-only facts are deleted (each has a
    Grpc counterpart that already asserted the same property independently), the three
    transport-agnostic facts move to Grpc mode, and the Dps fact keeps its own coverage.
  - MeshClusterClientFactoryTests deleted (its subject is gone).
  - MeshClusterClientBoundaryTests deleted here rather than in step 7: it drives
    DefaultMeshClusterClientFactory directly, so it cannot compile past this commit. Its
    replacement, MeshGrpcBoundaryTests, lands in step 7 and keeps the load-bearing assertion
    (two separate meshes, no cross-mesh Akka association).
This commit is contained in:
Joseph Doherty
2026-08-06 15:35:54 -04:00
parent 5510452662
commit 0db40a5e7f
6 changed files with 78 additions and 1108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
+46 -260
View File
@@ -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;
/// <summary>
/// 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
/// <see cref="MeshTransportOptions.Mode"/> — and forwards inbound <see cref="ApplyAck"/>s to the
/// deploy coordinator singleton.
/// every central→node command out — over DistributedPubSub (pair-local) or gRPC (cross-mesh),
/// per <see cref="MeshTransportOptions.Mode"/> — and forwards inbound <see cref="ApplyAck"/>s to
/// the deploy coordinator singleton.
/// </summary>
/// <remarks>
/// <para>
/// <b>Registered per node, NOT as a cluster singleton.</b> 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.
/// <b>Registered per node, NOT as a cluster singleton.</b> 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.
/// </para>
/// <para>
/// <b>One ClusterClient per application <c>Cluster</c> (Phase 6).</b> The fleet is now one
/// Akka mesh per cluster, so a single fleet-wide client's <c>SendToAll</c> would reach only
/// the one mesh whose receptionist answered. This actor therefore holds a client per cluster
/// and fans <c>SendToAll</c> across all of them. See <see cref="RebuildClient"/> — this is the
/// single most important thing to understand before changing this actor.
/// <b>The gRPC branch fans node by node over the contact load's target set.</b> That load
/// selects exactly the enabled, non-maintenance <c>ClusterNode</c> rows carrying a
/// <c>GrpcPort</c>, and the fan dials every one of them, one call per node. It is deliberately
/// NOT narrower: the node side has no <c>ClusterId</c> 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.
/// </para>
/// <para>
/// <b>The gRPC branch fans over the same recipient set, node by node</b>
/// (ClusterClient→gRPC Phase 1A). <c>SendToAll</c> 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 <c>ClusterId</c> 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.
/// </para>
/// <para>
/// <b>Fire-and-forget, on every transport.</b> The gRPC branch pipes per-node results back to
/// <b>Fire-and-forget, on both transports.</b> 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).
/// </para>
/// </remarks>
@@ -62,7 +51,6 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly MeshTransportOptions _options;
private readonly IMeshClusterClientFactory _clientFactory;
private readonly Func<IActorRef?> _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<string, (IActorRef Client, ImmutableHashSet<string> 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
/// <summary>Creates the props for this actor.</summary>
/// <param name="dbFactory">Factory for the config database holding <c>ClusterNode</c> rows.</param>
/// <param name="options">The bound mesh-transport options.</param>
/// <param name="clientFactory">Creates the ClusterClient; substituted in tests.</param>
/// <param name="coordinator">
/// Resolves the deploy-coordinator singleton proxy. A <see cref="Func{TResult}"/> rather than
/// an <see cref="IActorRef"/> so this actor never depends on registration order at wiring
@@ -129,9 +109,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
/// </param>
/// <param name="dialer">
/// Resolves the thing that places one command call on one node, under
/// <see cref="MeshTransportOptions.ModeGrpc"/>. Optional so the two older transports need
/// nothing new wired; a <see langword="null"/> dialer under <c>Grpc</c> is a mis-wiring and
/// is reported loudly rather than silently dropping every command.
/// <see cref="MeshTransportOptions.ModeGrpc"/>. Optional so the Dps transport needs nothing
/// new wired; a <see langword="null"/> dialer under <c>Grpc</c> is a mis-wiring and is
/// reported loudly rather than silently dropping every command.
/// </param>
/// <param name="pskProvider">
/// 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<OtOpcUaConfigDbContext> dbFactory,
MeshTransportOptions options,
IMeshClusterClientFactory clientFactory,
Func<IActorRef?> coordinator,
Func<IMeshCommandDialer?>? dialer = null,
Func<IMeshPskProvider?>? pskProvider = null) =>
Akka.Actor.Props.Create(() =>
new CentralCommunicationActor(
dbFactory, options, clientFactory, coordinator, dialer, pskProvider));
new CentralCommunicationActor(dbFactory, options, coordinator, dialer, pskProvider));
/// <summary>Initializes a new instance of the <see cref="CentralCommunicationActor"/> class.</summary>
/// <param name="dbFactory">Factory for the config database.</param>
/// <param name="options">The bound mesh-transport options.</param>
/// <param name="clientFactory">Creates the ClusterClient.</param>
/// <param name="coordinator">Lazily resolves the deploy-coordinator singleton proxy.</param>
/// <param name="dialer">Resolves the gRPC command dialer; invoked in <see cref="PreStart"/>.</param>
/// <param name="pskProvider">Resolves the PSK provider; invoked in <see cref="PreStart"/>.</param>
public CentralCommunicationActor(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
MeshTransportOptions options,
IMeshClusterClientFactory clientFactory,
Func<IActorRef?> coordinator,
Func<IMeshCommandDialer?>? dialer = null,
Func<IMeshPskProvider?>? 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
/// <inheritdoc />
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));
}
/// <summary>
@@ -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<string, List<string>>();
var malformed = new List<string>();
// 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<MeshNodeTarget>(rows.Count);
var portless = new List<string>();
@@ -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<string>();
list.Add(address);
}
else
{
malformed.Add($"{row.NodeId} ({row.ClusterId}) -> {address}");
}
}
var contactsByCluster = byCluster.ToDictionary(
kv => kv.Key, kv => (IReadOnlyList<string>)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");
}
}
/// <summary>
@@ -790,11 +642,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
/// </summary>
/// <param name="msg">The load result.</param>
/// <remarks>
/// 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.
/// </remarks>
private void AdoptGrpcTargets(ContactsLoaded msg)
{
@@ -837,7 +687,7 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
/// <b>Not gated on <see cref="MeshTransportOptions.Mode"/>, on purpose.</b> The gate feeds
/// <c>CentralCommandAuthInterceptor</c>, 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 <c>Grpc</c> would mean a dark-switch rollback to <c>ClusterClient</c> or
/// this on <c>Grpc</c> would mean a dark-switch rollback to
/// <c>Dps</c> 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();
}
/// <summary>Stops a cluster's existing client and creates a replacement for its new contact set.</summary>
/// <remarks>
/// <para>
/// <b>One client per application <c>Cluster</c> (delivered in Phase 6).</b> This actor now
/// holds one <c>ClusterClient</c> per cluster and <see cref="HandleMeshCommand"/> fans
/// <c>SendToAll</c> across all of them. Once the fleet is split into one Akka mesh per
/// cluster, a <c>ClusterClientReceptionist</c> serves only <i>its own</i> mesh, so a single
/// client's <c>SendToAll</c> 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.
/// </para>
/// <para>
/// Why this is not N× duplicate dispatch: each client dials only its own cluster's
/// receptionists, so a command is delivered <b>once per mesh</b>, 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.)
/// </para>
/// <para>
/// Corollary worth stating because it is easy to assume otherwise: <b>the contact set
/// does not scope delivery within a mesh.</b> 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 <i>dialling</i> (a node in maintenance may be switched off), not about who gets
/// the message.
/// </para>
/// </remarks>
/// <param name="clusterId">The application <c>Cluster</c> whose client is being (re)built.</param>
/// <param name="contacts">The receptionist addresses to dial for that cluster.</param>
private void RebuildClient(string clusterId, ImmutableHashSet<string> 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);
}
}
/// <summary>Internal tick asking for a contact-point refresh.</summary>
public sealed record RefreshContacts;
/// <summary>Result of a contact-point load, piped back from the DB read.</summary>
/// <param name="ContactsByCluster">
/// Receptionist addresses that parsed cleanly, grouped by the owning application
/// <c>Cluster</c>'s id. One <c>ClusterClient</c> is built per key.
/// </param>
/// <param name="Malformed">Rows that did not parse, described for the log.</param>
/// <param name="Targets">
/// The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a
/// <c>GrpcPort</c>. Built from the same rows as <paramref name="ContactsByCluster"/>.
/// <c>GrpcPort</c>.
/// </param>
/// <param name="PortlessNodes">
/// Node ids skipped because their row has no <c>GrpcPort</c>. 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.
/// </param>
public sealed record ContactsLoaded(
IReadOnlyDictionary<string, IReadOnlyList<string>> ContactsByCluster,
IReadOnlyList<string> Malformed,
IReadOnlyList<MeshNodeTarget> Targets,
IReadOnlyList<string> PortlessNodes,
IReadOnlyList<string> ClusterIds);
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshClusterClientFactory.cs
-67
View File
@@ -1,67 +0,0 @@
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
/// <summary>
/// Creates the <c>ClusterClient</c> central uses to reach driver nodes. Exists as a seam so
/// <c>CentralCommunicationActor</c> can be tested against a <c>TestProbe</c> without standing up
/// a second cluster.
/// </summary>
public interface IMeshClusterClientFactory
{
/// <summary>Creates a client for the given receptionist contact points.</summary>
/// <param name="system">The actor system to spawn the client in.</param>
/// <param name="clusterId">
/// The application <c>Cluster</c> 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.
/// </param>
/// <param name="contacts">Receptionist actor paths, one per reachable node.</param>
/// <returns>The new client's <see cref="IActorRef"/>.</returns>
IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet<ActorPath> contacts);
}
/// <inheritdoc />
public sealed class DefaultMeshClusterClientFactory : IMeshClusterClientFactory
{
/// <summary>
/// Per-incarnation generation counter, appended to the actor name.
/// </summary>
/// <remarks>
/// <c>Context.Stop</c> of the previous client is <b>asynchronous</b> — its actor name stays
/// reserved until termination completes — so recreating the same name while handling a single
/// message throws <see cref="InvalidActorNameException"/>. 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.
/// </remarks>
private long _generation;
/// <inheritdoc />
public IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet<ActorPath> 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);
}
/// <summary>
/// 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
/// <c>[A-Za-z0-9-]</c> becomes <c>-</c>. The trailing generation counter still guarantees
/// uniqueness, so a lossy sanitisation cannot cause a name collision.
/// </summary>
/// <param name="clusterId">The raw cluster id.</param>
/// <returns>An actor-name-safe rendering of the id.</returns>
private static string SanitizeForActorName(string clusterId)
{
var chars = clusterId.Select(c => char.IsAsciiLetterOrDigit(c) || c == '-' ? c : '-');
return new string(chars.ToArray());
}
}
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs
+5 -6
View File
@@ -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<IDbContextFactory<OtOpcUaConfigDbContext>>();
@@ -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<ConfigPublishCoordinatorKey>(out var coordinator)
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
+27 -247
View File
@@ -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";
/// <summary>Records every client the actor asks for, and hands back a probe in its place.</summary>
private sealed class RecordingClientFactory : IMeshClusterClientFactory
{
private readonly Func<TestProbe> _newProbe;
public RecordingClientFactory(Func<TestProbe> newProbe) => _newProbe = newProbe;
public List<ImmutableHashSet<ActorPath>> Calls { get; } = [];
public List<string> ClusterIds { get; } = [];
public List<TestProbe> Probes { get; } = [];
/// <summary>Probe created for a given cluster id, for per-cluster fan-out assertions.</summary>
public Dictionary<string, TestProbe> ProbeByCluster { get; } = [];
public IActorRef Create(ActorSystem system, string clusterId, ImmutableHashSet<ActorPath> 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<OtOpcUaConfigDbContext> 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();
}
/// <summary>
/// Seeds a cluster whose nodes carry a <c>GrpcPort</c> — the gRPC command plane's shape.
/// Kept separate from <see cref="SeedCluster"/> so the ClusterClient facts, which have no
/// opinion about the port, stay exactly as they were.
/// </summary>
/// <summary>Seeds a cluster whose nodes carry a <c>GrpcPort</c> — the command plane's shape.</summary>
private static void SeedGrpcCluster(
IDbContextFactory<OtOpcUaConfigDbContext> factory,
string clusterId,
@@ -145,15 +68,16 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
private IActorRef Spawn(
IDbContextFactory<OtOpcUaConfigDbContext> 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<SubscribeAck>(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<DispatchDeployment>(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<ClusterClient.SendToAll>(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<ClusterClient.SendToAll>(
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<SubscribeAck>(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<OtOpcUaConfigDbContext> 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));
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshClusterClientFactoryTests.cs
-82
View File
@@ -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;
/// <summary>
/// Guards the one non-obvious part of client creation: recreating a client with the same name
/// inside a single message handler throws, because <c>Context.Stop</c> is asynchronous and the
/// name stays reserved until termination completes.
/// </summary>
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<ActorPath> 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.
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs
-446
View File
@@ -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;
/// <summary>
/// Drives a real <c>ClusterClient</c> across two genuinely separate <see cref="ActorSystem"/>s —
/// the only test in the repo where the mesh boundary carries a message over the wire.
/// </summary>
/// <remarks>
/// <para>
/// The sister project substitutes an in-process relay that unwraps
/// <see cref="ClusterClient.Send"/> and forwards. That tests actor <i>logic</i> 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.
/// </para>
/// <para>
/// <b>Two clusters, one ActorSystem name, joined ONLY by ClusterClient.</b> The shared name
/// is mandatory — Akka.Remote matches on the full address, so a contact path naming
/// <c>otopcua</c> cannot reach a system called anything else. Each system seeds itself and
/// neither lists the other, and
/// <see cref="The_two_systems_are_separate_clusters_joined_only_by_the_client"/> 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.
/// </para>
/// <para>
/// <b>Scope limit.</b> 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 <c>MeshCommActorPathTests</c>
/// 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.
/// </para>
/// </remarks>
public sealed class MeshClusterClientBoundaryTests : IAsyncLifetime
{
private const string SharedSystemName = "otopcua";
private static readonly TimeSpan Arrival = TimeSpan.FromSeconds(30);
/// <summary>
/// How long the falsifiability control waits before concluding nothing arrived. Deliberately
/// shorter than <see cref="Arrival"/> but far longer than a successful delivery takes in
/// practice, so a pass means "the boundary is shut", not "the clock ran out first".
/// </summary>
private static readonly TimeSpan Silence = TimeSpan.FromSeconds(10);
/// <summary>
/// One factory for every client this class creates, because the generation counter that keeps
/// client actor names unique is <b>per factory instance</b>. 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
/// <c>CentralCommunicationActor</c>, so the real code path is safe.
/// </summary>
private readonly DefaultMeshClusterClientFactory _clientFactory = new();
private ActorSystem _central = null!;
private ActorSystem _node = null!;
private IActorRef _nodeComm = null!;
private IActorRef _centralClient = null!;
private TaskCompletionSource<object> _centralInbox = null!;
/// <inheritdoc />
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));
}
/// <inheritdoc />
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<RestartDriver>().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<ApplyAck>().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);
}
}
/// <summary>Registers a real node-comm actor with a system's receptionist so central can reach it.</summary>
/// <param name="system">The site system to register into.</param>
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<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext>
{
public OtOpcUaConfigDbContext CreateDbContext()
{
var opts = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
.UseInMemoryDatabase(dbName)
.Options;
return new OtOpcUaConfigDbContext(opts);
}
}
/// <summary>
/// Starts a single-member cluster on a dynamic port, optionally without the receptionist.
/// </summary>
/// <param name="withReceptionist">
/// <see langword="false"/> drops the receptionist provider from <c>akka.extensions</c>,
/// leaving DistributedPubSub in place so only the one variable changes.
/// </param>
/// <returns>The started system.</returns>
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");
}
/// <summary>
/// Re-sends until the inbox completes or the deadline passes.
/// </summary>
/// <remarks>
/// A ClusterClient establishes contact asynchronously and <c>buffer-size = 0</c> 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.
/// </remarks>
/// <param name="target">Actor to send to.</param>
/// <param name="message">Builds the message for each attempt.</param>
/// <param name="inbox">Completion source the receiving capture actor resolves.</param>
/// <param name="timeout">How long to keep trying.</param>
/// <returns>The delivered message, or <see langword="null"/> if none arrived.</returns>
private static async Task<object?> SendUntilDeliveredAsync(
IActorRef target,
Func<object> message,
TaskCompletionSource<object> 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<object> NewInbox() =>
new(TaskCreationOptions.RunContinuationsAsynchronously);
private static ImmutableHashSet<ActorPath> ReceptionistOf(ActorSystem system) =>
ImmutableHashSet.Create<ActorPath>(
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));
}
/// <summary>Completes an inbox with the first message it receives.</summary>
private sealed class CaptureActor : ReceiveActor
{
public static Props Props(TaskCompletionSource<object> sink) =>
Akka.Actor.Props.Create(() => new CaptureActor(sink));
public CaptureActor(TaskCompletionSource<object> sink) => ReceiveAny(m => sink.TrySetResult(m));
}
}