feat(mesh): CentralCommunicationActor — dark-switched fan-out, DB-sourced contacts
Phase 2 Task 3. Central-side end of the boundary: routes every central->node command out over DPS or ClusterClient per MeshTransport:Mode, and forwards inbound ApplyAcks to the deploy coordinator singleton (Forward, not Tell, so the coordinator sees the node rather than this relay). Three things here are load-bearing and easy to get wrong later: SendToAll, never Send. Today's DPS publish reaches EVERY DriverHostActor and the node side has no ClusterId or node filter to compensate -- scoping happens later, inside the artifact. ClusterClient.Send delivers to exactly ONE registered actor, so it would deploy to a single node while every other node silently kept its old config, and the deployment could still seal green. Sabotage-verified: swapping to Send reddens exactly that one test. Exactly ONE ClusterClient, fleet-wide. A receptionist serves its whole cluster, so SendToAll reaches every registered node-comm actor in the mesh regardless of which contact point was dialled. One client per application Cluster -- the shape Phase 6 wants -- would fan each command out once per cluster while the fleet is still one mesh: N x duplicate DispatchDeployment and ApplyAck. Marked TODO(Phase 6). 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, as it does under today's broadcast. The filter is about which receptionists are worth dialling, not about who gets the message. Stated in the code because the opposite is the natural assumption. Also carries the sister project's shipped fixes: per-row address parsing so one malformed ClusterNode cannot abort the refresh and leave the contact set half-built (regression test orders the bad row FIRST), the cache entry cleared before recreate so a failed create cannot leave commands routing into a stopping actor, and a Status.Failure handler so a DB outage is a Warning rather than a silent debug-level unhandled message. Two test-harness bugs found and fixed while writing this, both mine: SubscribeAck goes to the SENDER (so the mediator subscribe needed an explicit sender), and EventFilter matches case-INSENSITIVELY, so a "was NOT" filter also caught this actor's own "was not created" warning. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,33 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
|
||||
|
||||
/// <summary>
|
||||
/// Actor paths the two mesh comm actors register under. <b>These strings are the wire
|
||||
/// contract</b> of the central↔node boundary.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Declared once, here, rather than as literals on each side. The <c>deployments</c> /
|
||||
/// <c>deployment-acks</c> topic names are declared twice — once in
|
||||
/// <c>ConfigPublishCoordinator</c> and again in <c>DriverHostActor</c> — 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Nothing else pins these values, which is why <c>MeshCommActorPathTests</c> boots a real
|
||||
/// host and <c>Identify</c>-probes both paths.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class MeshPaths
|
||||
{
|
||||
/// <summary>Central-side comm actor — where a driver node sends its deployment acks.</summary>
|
||||
public const string CentralCommunication = "/user/central-communication";
|
||||
|
||||
/// <summary>Node-side comm actor — where central sends commands.</summary>
|
||||
public const string NodeCommunication = "/user/node-communication";
|
||||
|
||||
/// <summary>Actor name of <see cref="CentralCommunication"/> (the path's final segment).</summary>
|
||||
public const string CentralCommunicationName = "central-communication";
|
||||
|
||||
/// <summary>Actor name of <see cref="NodeCommunication"/> (the path's final segment).</summary>
|
||||
public const string NodeCommunicationName = "node-communication";
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Exactly one ClusterClient, fleet-wide.</b> See <see cref="RebuildClient"/> — this is
|
||||
/// the single most important thing to understand before changing this actor.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
private const string RefreshTimerKey = "mesh-contact-refresh";
|
||||
|
||||
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
|
||||
private readonly MeshTransportOptions _options;
|
||||
private readonly IMeshClusterClientFactory _clientFactory;
|
||||
private readonly Func<IActorRef?> _coordinator;
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
private readonly bool _useClusterClient;
|
||||
|
||||
private IActorRef? _client;
|
||||
private ImmutableHashSet<string> _currentContacts = ImmutableHashSet<string>.Empty;
|
||||
private CancellationTokenSource _lifecycle = new();
|
||||
|
||||
/// <summary>Gets the timer scheduler driving the periodic contact refresh.</summary>
|
||||
public ITimerScheduler Timers { get; set; } = null!;
|
||||
|
||||
/// <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
|
||||
/// time — it resolves on first use and caches.
|
||||
/// </param>
|
||||
/// <returns>The props.</returns>
|
||||
public static Props Props(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
||||
MeshTransportOptions options,
|
||||
IMeshClusterClientFactory clientFactory,
|
||||
Func<IActorRef?> coordinator) =>
|
||||
Akka.Actor.Props.Create(() =>
|
||||
new CentralCommunicationActor(dbFactory, options, clientFactory, coordinator));
|
||||
|
||||
/// <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>
|
||||
public CentralCommunicationActor(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
||||
MeshTransportOptions options,
|
||||
IMeshClusterClientFactory clientFactory,
|
||||
Func<IActorRef?> coordinator)
|
||||
{
|
||||
_dbFactory = dbFactory;
|
||||
_options = options;
|
||||
_clientFactory = clientFactory;
|
||||
_coordinator = coordinator;
|
||||
_useClusterClient = string.Equals(
|
||||
options.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
Receive<MeshCommand>(HandleMeshCommand);
|
||||
Receive<ApplyAck>(HandleApplyAck);
|
||||
Receive<RefreshContacts>(_ => LoadContactsFromDb());
|
||||
Receive<ContactsLoaded>(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<Status.Failure>(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<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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<string>(rows.Count);
|
||||
var malformed = new List<string>();
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>Stops any existing client and creates a replacement for the new contact set.</summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>TODO(Phase 6): one client per application <c>Cluster</c>.</b> Today this actor
|
||||
/// creates exactly <b>one</b> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// A <c>ClusterClientReceptionist</c> serves its entire cluster, so
|
||||
/// <c>SendToAll</c> reaches every registered node-comm actor in the mesh <i>regardless of
|
||||
/// which node's address was used as the contact point</i>. One client per Cluster would
|
||||
/// therefore fan each command out once per cluster — N× duplicate
|
||||
/// <c>DispatchDeployment</c> and N× duplicate <c>ApplyAck</c>. Phase 6 splits the meshes,
|
||||
/// at which point per-cluster clients become both correct and necessary.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Corollary worth stating because it is easy to assume otherwise: <b>the contact set
|
||||
/// does not scope delivery.</b> 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 <i>dialling</i>
|
||||
/// (a node in maintenance may be switched off), not about who gets the message.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="contacts">The receptionist addresses to dial.</param>
|
||||
private void RebuildClient(ImmutableHashSet<string> 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<string>.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");
|
||||
}
|
||||
}
|
||||
|
||||
/// <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="Contacts">Receptionist addresses that parsed cleanly.</param>
|
||||
/// <param name="Malformed">Rows that did not, described for the log.</param>
|
||||
public sealed record ContactsLoaded(
|
||||
IReadOnlyList<string> Contacts,
|
||||
IReadOnlyList<string> Malformed);
|
||||
}
|
||||
+271
@@ -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";
|
||||
|
||||
/// <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<TestProbe> Probes { get; } = [];
|
||||
|
||||
public IActorRef Create(ActorSystem system, ImmutableHashSet<ActorPath> 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<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();
|
||||
}
|
||||
|
||||
private IActorRef Spawn(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> 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<SubscribeAck>(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<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 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<ApplyAck>(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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user