diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs
new file mode 100644
index 00000000..308117f1
--- /dev/null
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs
@@ -0,0 +1,33 @@
+namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
+
+///
+/// Actor paths the two mesh comm actors register under. These strings are the wire
+/// contract of the central↔node boundary.
+///
+///
+///
+/// Declared once, here, rather than as literals on each side. The deployments /
+/// deployment-acks topic names are declared twice — once in
+/// ConfigPublishCoordinator and again in DriverHostActor — and agree only by
+/// convention; renaming one would compile, deploy, and deliver nothing. A ClusterClient path
+/// has the same failure mode and no compiler to catch it, so the constant is shared.
+///
+///
+/// Nothing else pins these values, which is why MeshCommActorPathTests boots a real
+/// host and Identify-probes both paths.
+///
+///
+public static class MeshPaths
+{
+ /// Central-side comm actor — where a driver node sends its deployment acks.
+ public const string CentralCommunication = "/user/central-communication";
+
+ /// Node-side comm actor — where central sends commands.
+ public const string NodeCommunication = "/user/node-communication";
+
+ /// Actor name of (the path's final segment).
+ public const string CentralCommunicationName = "central-communication";
+
+ /// Actor name of (the path's final segment).
+ public const string NodeCommunicationName = "node-communication";
+}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
new file mode 100644
index 00000000..333087e6
--- /dev/null
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/CentralCommunicationActor.cs
@@ -0,0 +1,308 @@
+using System.Collections.Immutable;
+using Akka.Actor;
+using Akka.Cluster.Tools.Client;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.Event;
+using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.OtOpcUa.Cluster;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+
+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 or ClusterClient, 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.
+///
+///
+/// Exactly one ClusterClient, fleet-wide. See — this is
+/// the single most important thing to understand before changing this actor.
+///
+///
+public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
+{
+ private const string RefreshTimerKey = "mesh-contact-refresh";
+
+ private readonly IDbContextFactory _dbFactory;
+ private readonly MeshTransportOptions _options;
+ private readonly IMeshClusterClientFactory _clientFactory;
+ private readonly Func _coordinator;
+ private readonly ILoggingAdapter _log = Context.GetLogger();
+ private readonly bool _useClusterClient;
+
+ private IActorRef? _client;
+ private ImmutableHashSet _currentContacts = ImmutableHashSet.Empty;
+ private CancellationTokenSource _lifecycle = new();
+
+ /// Gets the timer scheduler driving the periodic contact refresh.
+ public ITimerScheduler Timers { get; set; } = null!;
+
+ /// 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
+ /// time — it resolves on first use and caches.
+ ///
+ /// The props.
+ public static Props Props(
+ IDbContextFactory dbFactory,
+ MeshTransportOptions options,
+ IMeshClusterClientFactory clientFactory,
+ Func coordinator) =>
+ Akka.Actor.Props.Create(() =>
+ new CentralCommunicationActor(dbFactory, options, clientFactory, coordinator));
+
+ /// 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.
+ public CentralCommunicationActor(
+ IDbContextFactory dbFactory,
+ MeshTransportOptions options,
+ IMeshClusterClientFactory clientFactory,
+ Func coordinator)
+ {
+ _dbFactory = dbFactory;
+ _options = options;
+ _clientFactory = clientFactory;
+ _coordinator = coordinator;
+ _useClusterClient = string.Equals(
+ options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
+
+ Receive(HandleMeshCommand);
+ Receive(HandleApplyAck);
+ Receive(_ => LoadContactsFromDb());
+ Receive(HandleContactsLoaded);
+
+ // A faulted LoadContactsFromDb task is piped here as a Status.Failure. Without this handler
+ // the failure is an unhandled message (debug level only) and the refresh fails silently —
+ // an operator cannot distinguish "no nodes configured" from "the database is down", and the
+ // contact set silently freezes at whatever it last held.
+ Receive(f => _log.Warning(
+ f.Cause,
+ "Failed to load ClusterNode contact points; the mesh ClusterClient contact set was NOT "
+ + "refreshed and may be stale or empty. Commands may be dropped until the next refresh"));
+
+ Receive(_ => { /* DPS subscribe confirmation */ });
+ }
+
+ ///
+ protected override void PreStart()
+ {
+ if (!_useClusterClient)
+ {
+ _log.Info(
+ "Mesh transport is {Mode}: commands are published on DistributedPubSub and no "
+ + "ClusterClient is created", _options.Mode);
+ return;
+ }
+
+ Timers.StartPeriodicTimer(
+ RefreshTimerKey,
+ new RefreshContacts(),
+ TimeSpan.Zero,
+ TimeSpan.FromSeconds(_options.ContactRefreshSeconds));
+ }
+
+ ///
+ protected override void PostStop()
+ {
+ _lifecycle.Cancel();
+ _lifecycle.Dispose();
+ }
+
+ private void HandleMeshCommand(MeshCommand cmd)
+ {
+ if (!_useClusterClient)
+ {
+ DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(cmd.Topic, cmd.Message));
+ return;
+ }
+
+ if (_client is null)
+ {
+ // 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; the "
+ + "command is not buffered and will not be retried",
+ cmd.Message.GetType().Name);
+ return;
+ }
+
+ // SendToAll, NEVER Send. Today's DPS publish reaches EVERY DriverHostActor — there is no
+ // ClusterId or node filter on the node side at all; scoping happens later, inside the
+ // artifact (DeploymentArtifact.ResolveClusterScope). ClusterClient.Send delivers to exactly
+ // ONE registered actor, which would deploy to a single node of the fleet while every other
+ // node silently kept its old configuration — and the deployment would still seal green if
+ // the ack set happened to be satisfied.
+ _client.Tell(new ClusterClient.SendToAll(MeshPaths.NodeCommunication, cmd.Message));
+ }
+
+ private void HandleApplyAck(ApplyAck ack)
+ {
+ var coordinator = _coordinator();
+ if (coordinator is null)
+ {
+ _log.Warning(
+ "Received ApplyAck for {DeploymentId} from {NodeId} but the deploy coordinator is "
+ + "not resolvable on this node; the ack is lost and the deployment will time out "
+ + "naming a node that applied successfully",
+ ack.DeploymentId, ack.NodeId);
+ return;
+ }
+
+ // Forward, not Tell: preserves the original sender across the hop so the coordinator sees
+ // the node rather than this relay.
+ coordinator.Forward(ack);
+ }
+
+ private void LoadContactsFromDb()
+ {
+ var self = Self;
+ CancellationToken ct;
+ try
+ {
+ ct = _lifecycle.Token;
+ }
+ catch (ObjectDisposedException)
+ {
+ 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
+ // block every command dispatch behind a slow or unreachable SQL Server.
+ Task.Run(async () =>
+ {
+ await using var db = await dbFactory.CreateDbContextAsync(ct).ConfigureAwait(false);
+ var rows = await db.ClusterNodes
+ .AsNoTracking()
+ .Where(n => n.Enabled && !n.MaintenanceMode)
+ .Select(n => new { n.NodeId, n.Host, n.AkkaPort })
+ .ToListAsync(ct)
+ .ConfigureAwait(false);
+
+ var contacts = new List(rows.Count);
+ var malformed = new List();
+ foreach (var row in rows)
+ {
+ 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 _)) contacts.Add(address);
+ else malformed.Add($"{row.NodeId} -> {address}");
+ }
+
+ return new ContactsLoaded(contacts, malformed);
+ }, ct).PipeTo(self);
+ }
+
+ private void HandleContactsLoaded(ContactsLoaded msg)
+ {
+ 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 contacts = msg.Contacts.ToImmutableHashSet();
+
+ if (contacts.IsEmpty)
+ {
+ _log.Warning(
+ "No usable ClusterNode contact points; the mesh ClusterClient was not created and "
+ + "every central→node command will be dropped until a refresh finds one");
+ return;
+ }
+
+ if (_client is not null && _currentContacts.SetEquals(contacts)) return;
+
+ RebuildClient(contacts);
+ }
+
+ /// Stops any existing client and creates a replacement for the new contact set.
+ ///
+ ///
+ /// TODO(Phase 6): one client per application Cluster. Today this actor
+ /// creates exactly one client whose contacts are the whole fleet, and that is not
+ /// a compromise — it is the only correct shape while the fleet is a single Akka mesh.
+ ///
+ ///
+ /// A ClusterClientReceptionist serves its entire cluster, so
+ /// SendToAll reaches every registered node-comm actor in the mesh regardless of
+ /// which node's address was used as the contact point. One client per Cluster would
+ /// therefore fan each command out once per cluster — N× duplicate
+ /// DispatchDeployment and N× duplicate ApplyAck. Phase 6 splits the meshes,
+ /// at which point per-cluster clients become both correct and necessary.
+ ///
+ ///
+ /// Corollary worth stating because it is easy to assume otherwise: the contact set
+ /// does not scope delivery. Excluding a maintenance-mode node from the contacts does
+ /// not stop it receiving commands — it still receives them, exactly as it does under
+ /// today's DPS broadcast. The filter is about which receptionists are worth dialling
+ /// (a node in maintenance may be switched off), not about who gets the message.
+ ///
+ ///
+ /// The receptionist addresses to dial.
+ private void RebuildClient(ImmutableHashSet contacts)
+ {
+ if (_client is not null)
+ {
+ _log.Info("Mesh contact set changed; replacing the ClusterClient");
+ Context.Stop(_client);
+
+ // Cleared now: if the create below throws, a stale ref would route commands into a
+ // stopping actor and they would vanish without a warning.
+ _client = null;
+ _currentContacts = ImmutableHashSet.Empty;
+ }
+
+ try
+ {
+ var paths = contacts.Select(ActorPath.Parse).ToImmutableHashSet();
+ _client = _clientFactory.Create(Context.System, paths);
+ _currentContacts = contacts;
+ _log.Info("Mesh ClusterClient created with {Count} contact point(s)", contacts.Count);
+ }
+ catch (Exception ex)
+ {
+ _log.Error(ex,
+ "Failed to create the mesh ClusterClient; central→node commands are dropped until "
+ + "the next contact refresh succeeds");
+ }
+ }
+
+ /// 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.
+ /// Rows that did not, described for the log.
+ public sealed record ContactsLoaded(
+ IReadOnlyList Contacts,
+ IReadOnlyList Malformed);
+}
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
new file mode 100644
index 00000000..d2d0fc7a
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
@@ -0,0 +1,271 @@
+using System.Collections.Immutable;
+using Akka.Actor;
+using Akka.Cluster.Tools.Client;
+using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.TestKit;
+using Microsoft.EntityFrameworkCore;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Cluster;
+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.Configuration.Enums;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication;
+
+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 Probes { get; } = [];
+
+ public IActorRef Create(ActorSystem system, ImmutableHashSet contacts)
+ {
+ Calls.Add(contacts);
+ var probe = _newProbe();
+ Probes.Add(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 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();
+ }
+
+ private IActorRef Spawn(
+ IDbContextFactory factory,
+ MeshTransportOptions options,
+ IMeshClusterClientFactory clientFactory,
+ IActorRef? coordinator = null) =>
+ Sys.ActorOf(CentralCommunicationActor.Props(factory, options, clientFactory, () => coordinator));
+
+ [Fact]
+ public void Dps_mode_publishes_on_the_carried_topic_and_creates_no_client()
+ {
+ 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.
+ DistributedPubSub.Get(Sys).Mediator.Tell(
+ new Subscribe(DeploymentsTopic, mediatorProbe.Ref), mediatorProbe.Ref);
+ mediatorProbe.ExpectMsg(TimeSpan.FromSeconds(5));
+
+ var actor = Spawn(factory, new MeshTransportOptions(), clients);
+ 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 Exactly_one_client_is_created_for_a_multi_cluster_fleet()
+ {
+ // A receptionist serves its whole mesh, so one client per Cluster would fan each command out
+ // once per cluster while the fleet is still a single mesh — N x duplicate deployments.
+ var factory = NewInMemoryDbFactory();
+ SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
+ SeedCluster(factory, "C2", ("host-b:4053", "host-b", 4053, true, false));
+ var clients = new RecordingClientFactory(() => CreateTestProbe());
+
+ Spawn(factory, ClusterClientMode(), clients);
+ AwaitCondition(() => clients.Calls.Count == 1, TimeSpan.FromSeconds(5));
+
+ // ...and that one client dials BOTH clusters' nodes.
+ clients.Calls[0].Count.ShouldBe(2);
+ // Give a second refresh a chance to (wrongly) create another.
+ ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+ clients.Calls.Count.ShouldBe(1);
+ }
+
+ [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 ack = new ApplyAck(
+ new DeploymentId(Guid.NewGuid()),
+ new NodeId("host-a:4053"),
+ ApplyAckOutcome.Applied,
+ null,
+ new CorrelationId(Guid.NewGuid()));
+
+ actor.Tell(ack, node.Ref);
+
+ coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).ShouldBe(ack);
+ // Forward, not Tell: the coordinator must see the node, not this relay.
+ coordinator.LastSender.ShouldBe(node.Ref);
+ }
+
+ [Fact]
+ 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 ack = new ApplyAck(
+ new DeploymentId(Guid.NewGuid()),
+ new NodeId("host-a:4053"),
+ ApplyAckOutcome.Applied,
+ null,
+ new CorrelationId(Guid.NewGuid()));
+
+ EventFilter.Warning(contains: "not resolvable").ExpectOne(TimeSpan.FromSeconds(5), () =>
+ actor.Tell(ack));
+ }
+
+ [Fact]
+ 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);
+
+ // 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.
+ EventFilter.Warning(contains: "contact set was NOT refreshed").ExpectOne(
+ TimeSpan.FromSeconds(5),
+ () => actor.Tell(new Status.Failure(new InvalidOperationException("db down"))));
+
+ // Still serving: a transient DB outage must not take the command boundary down with it.
+ actor.Tell(new MeshCommand(DeploymentsTopic, "anything"));
+ ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+ }
+}