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

test(mesh): real two-ActorSystem ClusterClient boundary test with falsifiability control

Browse Source 
Two genuinely separate single-member clusters sharing the ActorSystem name and
joined only by ClusterClient, so the mesh boundary is exercised over the wire
rather than through an in-process relay: a relay cannot fail for a missing
receptionist extension, an unregistered service, a malformed contact path or a
serialization break, which are the Phase 2 failure modes.

Covers both legs plus the control:
  * central SendToAll -> node comm actor -> node EventStream
  * node ApplyAck -> central comm actor, through the node's own client
  * a node without the receptionist extension receives nothing

Sabotage-verified. Suppressing the node-side EventStream publish and the
outbound ack Tell reddens one leg each. Restoring the receptionist and the
service registration on the control node reddens the control, which is what
proves the control's send is live rather than silently misaddressed.

The control removes the extension AND the RegisterService call, since resolving
the receptionist to register would materialise the extension whose absence is
the point; it falsifies "delivery happens without a receptionist boundary", not
the extension line alone. Recorded in the test.

Also corrects MeshCommActorPathTests: with one node per side, per-node and
singleton registration are indistinguishable here, so the property the dropped
Task 6 assertion covered belongs to live-gate step 8, not to this class.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-22 17:32:19 -04:00
parent d246a7e44d
commit 5468b79d19
2 changed files with 308 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshClusterClientBoundaryTests.cs
+302
View File
@@ -0,0 +1,302 @@
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.Client;
using Akka.Configuration;
using Shouldly;
using Xunit;
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.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, ReceptionistOf(_central));
_nodeComm = _node.ActorOf(
NodeCommunicationActor.Props(nodeToCentral), MeshPaths.NodeCommunicationName);
ClusterClientReceptionist.Get(_node).RegisterService(_nodeComm);
_centralClient = _clientFactory.Create(_central, 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, 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);
}
}
/// <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));
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs
+6 -2
View File
@@ -57,6 +57,10 @@ public class MeshCommActorPathTests
//
// A sabotage that re-registered CentralCommunicationActor via WithSingleton left both green,
// which is what exposed them. Rather than ship an assertion I could not make fail, the property
// is left to MeshClusterClientBoundaryTests: a ClusterClient send only arrives if the target is
// actually registered with the receptionist, which is the thing that matters.
// is left to live-gate step 8 (stop the OLDEST central; the survivor's comm actor must still
// receive a node ack, which a singleton hosted on the stopped node cannot).
//
// NOT MeshClusterClientBoundaryTests: that class proves a send only arrives if the target is
// registered with the receptionist, which is necessary but not discriminating — with one node
// per side, per-node and singleton registration look identical.
}