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

feat(mesh): gRPC command fan-out behind Mode=Grpc

Browse Source 
MeshGrpcCommandDialer places one NodeCommandService.Execute call on one
node: a GrpcChannel cached per endpoint (the GrpcDeploymentArtifactFetcher
shape), the TARGET node's cluster's bearer plus x-otopcua-cluster naming
that cluster, and a per-call deadline linked to the caller's token. A
cluster with no resolvable key is refused locally with Unauthenticated —
there is no arm on which a command leaves central unauthenticated, even
one the far side would have rejected.

CentralCommunicationActor gains the third mode branch. The contact load
now also projects GrpcPort and, from the SAME rows in the SAME pass,
builds the dispatch set — which is what makes 'the gRPC fan reaches
exactly the SendToAll recipient set' structural rather than two queries
that have to keep agreeing. A row with a null GrpcPort is skipped
warn-once (the TelemetryNodeSource precedent). Per-node results are
piped back for LOGGING ONLY: named node, gRPC status, dropped. No retry,
no queue.

Every dial is scheduled on the thread pool rather than called from the
receive. Found by review and pinned red-first: an async method runs
synchronously to its first await, and the dialer's pre-await work is
TryGetPsk, which blocks on the secret store — bounded but seconds, and
repeated after every refresh since the refresh drops the cache. Dialing
inline parked the actor's mailbox, so deployment acks queued behind a
slow node; the regression test tells an ApplyAck while a dial is held
and fails without the hop.

The contact refresh also wires the Phase 1A register: it publishes the
loaded cluster ids as the PSK provider's known-cluster set and drops the
key cache, so a rotated key takes effect within one ContactRefreshSeconds.
Neither happens under Dps or ClusterClient.

Both existing mode branches are behaviourally untouched and every
existing fact passes unmodified. The one edit to shared text: the
Status.Failure warning drops the word 'ClusterClient', since that handler
now serves all three transports.

ControlPlane.Tests 205 -> 230, Host.Tests 92, Cluster.Tests 184, 0 warnings.
This commit is contained in:
Joseph Doherty
2026-08-06 05:58:21 -04:00
parent 14d21496fe
commit 6a86201105
6 changed files with 1428 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs
+561
View File
@@ -1,4 +1,5 @@
using System.Collections.Immutable;
using System.Diagnostics.CodeAnalysis;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Cluster.Tools.PublishSubscribe;
@@ -9,6 +10,8 @@ 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.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
@@ -56,6 +59,13 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
ContactRefreshSeconds = refreshSeconds,
};
private static MeshTransportOptions GrpcMode(int refreshSeconds = 60) => new()
{
Mode = MeshTransportOptions.ModeGrpc,
ContactRefreshSeconds = refreshSeconds,
DispatchTimeoutSeconds = 10,
};
private static void SeedCluster(
IDbContextFactory<OtOpcUaConfigDbContext> factory,
string clusterId,
@@ -92,6 +102,46 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
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>
private static void SeedGrpcCluster(
IDbContextFactory<OtOpcUaConfigDbContext> factory,
string clusterId,
params (string NodeId, string Host, int? GrpcPort, bool Enabled, bool Maintenance)[] nodes)
{
using var db = factory.CreateDbContext();
db.ServerClusters.Add(new ServerCluster
{
ClusterId = clusterId,
Name = clusterId,
Enterprise = "ZB",
Site = "Site",
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 = 4053,
GrpcPort = n.GrpcPort,
ApplicationUri = $"urn:{n.NodeId}",
Enabled = n.Enabled,
MaintenanceMode = n.Maintenance,
CreatedBy = "test",
});
}
db.SaveChanges();
}
private IActorRef Spawn(
IDbContextFactory<OtOpcUaConfigDbContext> factory,
MeshTransportOptions options,
@@ -306,4 +356,515 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
actor.Tell(new MeshCommand(DeploymentsTopic, "anything"));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
// ------------------------------------------------------------------ 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.
[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.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", 4056, true, false),
("host-b:4053", "host-b", 4056, true, false));
SeedGrpcCluster(factory, "C2", ("host-c:4053", "host-c", 4056, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 3, TimeSpan.FromSeconds(5));
rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
new[] { "http://host-a:4056", "http://host-b:4056", "http://host-c:4056" }, ignoreOrder: true);
// Each node is dialled with ITS OWN cluster's id — that is what selects the pre-shared key,
// so getting it wrong turns every command into a PermissionDenied at the node.
rig.Dialer.Calls.Single(c => c.Endpoint.Contains("host-c")).ClusterId.ShouldBe("C2");
rig.Dialer.Calls.Where(c => !c.Endpoint.Contains("host-c")).ShouldAllBe(c => c.ClusterId == "C1");
}
[Fact]
public void Grpc_mode_creates_no_ClusterClient_and_publishes_nothing_on_pubsub()
{
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);
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.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", 4056, true, false),
("host-b:4053", "host-b", 4056, false, false),
("host-c:4053", "host-c", 4056, true, true));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
rig.Dialer.Calls[0].Endpoint.ShouldContain("host-a");
// And no late arrival from the two excluded rows.
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
rig.Dialer.Calls.Count.ShouldBe(1);
}
[Fact]
public void A_node_with_no_GrpcPort_is_skipped_with_a_warning_and_the_others_are_still_dialled()
{
// Mirrors TelemetryNodeSource: a row with no port exposes no endpoint yet. It must cost that
// node its commands, never the whole fan.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", null, true, false),
("host-b:4053", "host-b", 4056, true, false));
GrpcRig? rig = null;
EventFilter.Warning(contains: "has no GrpcPort").ExpectOne(
TimeSpan.FromSeconds(5), () => rig = SpawnGrpcRig(factory));
rig.ShouldNotBeNull();
AwaitContactsLoaded(rig);
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
rig.Dialer.Calls[0].Endpoint.ShouldBe("http://host-b:4056");
}
[Fact]
public void One_failing_node_does_not_stop_the_fan_and_is_logged_by_node_id()
{
// Fire-and-forget: a failure is logged and DROPPED. No retry, no queue, and above all no
// abort of the remaining nodes — the deploy path's apply deadline is the backstop.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", 4056, true, false),
("host-b:4053", "host-b", 4056, true, false));
var dialer = new RecordingDialer
{
FailEndpointsContaining = "host-a",
Fault = new Grpc.Core.RpcException(
new Grpc.Core.Status(Grpc.Core.StatusCode.Unavailable, "node down")),
};
var rig = SpawnGrpcRig(factory, dialer);
AwaitContactsLoaded(rig);
// The node id, not the endpoint: an operator reading the log is looking at ClusterNode rows.
EventFilter.Warning(contains: "host-a:4053").ExpectOne(TimeSpan.FromSeconds(5), () =>
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
// Both were attempted; the healthy one is unaffected by its neighbour's failure.
AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
// Still serving after a per-node failure — no supervision restart, no state loss.
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 4, TimeSpan.FromSeconds(5));
}
[Fact]
public void A_psk_miss_costs_only_that_node_and_never_becomes_a_fan_level_abort()
{
// The dialer refuses a keyless cluster locally, so it arrives here as an ordinary per-node
// failure. The fan-level assertion is that the other cluster is still dialled.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", 4056, true, false));
var dialer = new RecordingDialer
{
FailClusters = ["C2"],
Fault = new Grpc.Core.RpcException(
new Grpc.Core.Status(Grpc.Core.StatusCode.Unauthenticated, "no key")),
};
var rig = SpawnGrpcRig(factory, dialer);
AwaitContactsLoaded(rig);
EventFilter.Warning(contains: "Unauthenticated").ExpectOne(TimeSpan.FromSeconds(5), () =>
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
}
[Fact]
public void A_command_with_no_grpc_target_is_dropped_with_a_warning()
{
var factory = NewInMemoryDbFactory();
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
EventFilter.Warning(contains: "dropping").ExpectOne(TimeSpan.FromSeconds(5), () =>
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
rig.Dialer.Calls.ShouldBeEmpty();
}
[Fact]
public void A_payload_that_is_not_a_mesh_command_record_is_dropped_with_an_error()
{
// The mapper refuses to encode an unknown payload rather than shipping something the node
// cannot decode. That refusal must cost the command, not the actor.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
EventFilter.Error(contains: "Cannot encode").ExpectOne(TimeSpan.FromSeconds(5), () =>
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, "not a command record")));
rig.Dialer.Calls.ShouldBeEmpty();
// Still serving.
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 1, TimeSpan.FromSeconds(5));
}
[Fact]
public void The_envelope_carries_the_commands_own_correlation_and_the_whole_fan_shares_it()
{
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", 4056, true, false),
("host-b:4053", "host-b", 4056, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
var correlationId = new CorrelationId(Guid.NewGuid());
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, new DispatchDeployment(
new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), correlationId)));
AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
rig.Dialer.Calls.ShouldAllBe(c => c.Envelope.CorrelationId == correlationId.ToString());
// One envelope object, encoded once and fanned — not re-encoded per node.
rig.Dialer.Calls.Select(c => c.Envelope).Distinct().Count().ShouldBe(1);
}
[Fact]
public void An_alarm_command_gets_one_minted_transport_correlation_shared_by_the_whole_fan()
{
// AlarmCommand is the one record with no correlation member of its own, so the envelope's is
// the only handle an operator has on it. Minting per fan rather than per node is what makes
// "this alarm command" one traceable event instead of N unrelated ones.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(
factory,
"C1",
("host-a:4053", "host-a", 4056, true, false),
("host-b:4053", "host-b", 4056, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
rig.Actor.Tell(new MeshCommand(
"alarms", new AlarmCommand("alarm-1", "Acknowledge", "operator", null, null)));
AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
rig.Dialer.Calls.Select(c => c.Envelope.CorrelationId).Distinct().Count().ShouldBe(1);
rig.Dialer.Calls[0].Envelope.CorrelationId.ShouldNotBeNullOrWhiteSpace();
rig.Dialer.Calls[0].Envelope.AlarmCommand.AlarmId.ShouldBe("alarm-1");
}
[Fact]
public void The_contact_refresh_publishes_the_known_cluster_set_and_drops_cached_psks()
{
// Phase 1A security register item 1, wired: the gate is only a control if something keeps it
// current, and the refresh is the moment central's picture of the fleet is up to date. The
// InvalidateAll on the same tick is what makes a rotated key take effect within one refresh
// instead of requiring a central restart.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
SeedGrpcCluster(factory, "C2", ("host-b:4053", "host-b", null, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
// C2 is included even though its only node has no GrpcPort: the gate is about which clusters
// central KNOWS, not which it can currently dial. Excluding it would deny that cluster's acks.
rig.Psk.KnownClusterPublications[0].ShouldBe(new[] { "C1", "C2" }, ignoreOrder: true);
rig.Psk.InvalidateAllCount.ShouldBeGreaterThanOrEqualTo(1);
}
[Fact]
public void A_slow_dial_does_not_park_the_actors_mailbox()
{
// The dialer's first act is a PSK lookup, which blocks on the secret store — bounded, but
// seconds, and repeated after every refresh because the refresh drops the cache. An async
// method runs synchronously to its first await, so dialing straight from the receive would
// stall this actor for that long PER NODE, and deployment acks queue behind that mailbox.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
var coordinator = CreateTestProbe();
var dialer = new RecordingDialer { BlockBeforeFirstAwait = true };
var rig = SpawnGrpcRig(factory, dialer, coordinator: coordinator.Ref);
AwaitContactsLoaded(rig);
try
{
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
// The ack must be handled while the dial is still parked.
var ack = new ApplyAck(
new DeploymentId(Guid.NewGuid()),
new NodeId("host-a:4053"),
ApplyAckOutcome.Applied,
null,
new CorrelationId(Guid.NewGuid()));
rig.Actor.Tell(ack);
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).ShouldBe(ack);
}
finally
{
dialer.Release();
}
}
[Fact]
public void A_node_added_after_start_joins_the_fan_on_the_next_refresh()
{
// The refresh is not decoration: operators add and retire nodes, and central must pick them
// up without a restart. Under Grpc that is the ONLY thing keeping the dispatch set current —
// there is no receptionist to notice a new registration on its own.
//
// Driven by telling the actor its own RefreshContacts tick rather than by shortening
// ContactRefreshSeconds: a one-second timer would run for the whole fact, adding DB load to
// every test sharing the runner, and would make this assertion depend on wall-clock timing.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
var rig = SpawnGrpcRig(factory);
AwaitContactsLoaded(rig);
using (var db = factory.CreateDbContext())
{
db.ClusterNodes.Add(new ClusterNode
{
NodeId = "host-b:4053",
ClusterId = "C1",
Host = "host-b",
AkkaPort = 4053,
GrpcPort = 4056,
ApplicationUri = "urn:host-b",
Enabled = true,
CreatedBy = "test",
});
db.SaveChanges();
}
rig.Actor.Tell(new CentralCommunicationActor.RefreshContacts());
AwaitCondition(
() => rig.Psk.KnownClusterPublications.Count >= 2, TimeSpan.FromSeconds(5));
rig.Actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch()));
AwaitCondition(() => rig.Dialer.Calls.Count == 2, TimeSpan.FromSeconds(5));
rig.Dialer.Calls.Select(c => c.Endpoint).ShouldBe(
new[] { "http://host-a:4056", "http://host-b:4056" }, ignoreOrder: true);
}
[Fact]
public void The_known_cluster_set_is_not_published_by_the_other_two_modes()
{
// The gate is a gRPC-command-plane control. Publishing it under Dps or ClusterClient would
// start refusing acks on a transport whose ack path never consults the provider at all —
// and, worse, would do so from a code path those modes' tests do not cover.
var factory = NewInMemoryDbFactory();
SeedCluster(factory, "C1", ("host-a:4053", "host-a", 4053, true, false));
var psk = new RecordingPskProvider();
Sys.ActorOf(CentralCommunicationActor.Props(
factory,
ClusterClientMode(),
new RecordingClientFactory(() => CreateTestProbe()),
() => null,
dialer: null,
pskProvider: psk));
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
psk.KnownClusterPublications.ShouldBeEmpty();
psk.InvalidateAllCount.ShouldBe(0);
}
[Fact]
public void Grpc_mode_with_no_dialer_wired_drops_the_command_with_an_error()
{
// Mis-wiring, not misconfiguration: Mode=Grpc validates fine but the DI graph did not supply
// a dialer. Silence here would be a fleet-wide command outage with no symptom until a deploy
// stalled at its apply deadline two minutes later.
var factory = NewInMemoryDbFactory();
SeedGrpcCluster(factory, "C1", ("host-a:4053", "host-a", 4056, true, false));
var actor = Sys.ActorOf(CentralCommunicationActor.Props(
factory,
GrpcMode(),
new RecordingClientFactory(() => CreateTestProbe()),
() => null,
dialer: null,
pskProvider: null));
// Filtered on the DISPATCH message's exact phrase: PreStart logs its own error about the same
// mis-wiring, and a looser filter races it into a second match.
EventFilter.Error(contains: "no command dialer is wired").ExpectOne(
TimeSpan.FromSeconds(5),
() => actor.Tell(new MeshCommand(DeploymentsTopic, NewDispatch())));
}
private static DispatchDeployment NewDispatch() => new(
new DeploymentId(Guid.NewGuid()), new RevisionHash("abc"), new CorrelationId(Guid.NewGuid()));
/// <summary>A spawned gRPC-mode actor plus the two seams the tests observe it through.</summary>
private sealed record GrpcRig(IActorRef Actor, RecordingDialer Dialer, RecordingPskProvider Psk);
private GrpcRig SpawnGrpcRig(
IDbContextFactory<OtOpcUaConfigDbContext> factory,
RecordingDialer? dialer = null,
IMeshClusterClientFactory? clientFactory = null,
IActorRef? coordinator = null)
{
dialer ??= new RecordingDialer();
var psk = new RecordingPskProvider();
var actor = Sys.ActorOf(CentralCommunicationActor.Props(
factory,
GrpcMode(),
clientFactory ?? new RecordingClientFactory(() => CreateTestProbe()),
() => coordinator,
dialer,
psk));
return new GrpcRig(actor, dialer, psk);
}
/// <summary>
/// Waits until the actor has applied a contact load. The known-cluster publication is the
/// signal because it happens in the same handler, immediately after the target set is
/// adopted — and unlike the target set itself it is observable from outside the actor. A
/// command told before this point would be dropped as "no targets" and never retried, which
/// is correct behaviour and a flaky test.
/// </summary>
private void AwaitContactsLoaded(GrpcRig rig) =>
AwaitCondition(
() => rig.Psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));
/// <summary>Records every dial and answers however the test asks; never opens a socket.</summary>
private sealed class RecordingDialer : IMeshCommandDialer
{
private readonly object _gate = new();
private readonly List<Dial> _calls = [];
private readonly ManualResetEventSlim _released = new(false);
/// <summary>
/// When true, every dial blocks its calling thread synchronously until
/// <see cref="Release"/>. Models the dialer's pre-await PSK lookup.
/// </summary>
public bool BlockBeforeFirstAwait { get; set; }
/// <summary>Unblocks every dial parked by <see cref="BlockBeforeFirstAwait"/>.</summary>
public void Release() => _released.Set();
/// <summary>Every dial made. Copied under the lock — the fan runs on thread-pool threads.</summary>
public IReadOnlyList<Dial> Calls
{
get { lock (_gate) return _calls.ToList(); }
}
/// <summary>Endpoints whose substring match makes the dial fault.</summary>
public string? FailEndpointsContaining { get; set; }
/// <summary>Cluster ids whose dials fault.</summary>
public string[] FailClusters { get; set; } = [];
/// <summary>
/// The fault raised for a matching dial. Fully qualified because this file's bare
/// <c>Status</c> is Akka's — an existing fact drives the actor's <c>Status.Failure</c>
/// handler, and importing Grpc.Core here would make that reference ambiguous.
/// </summary>
public Grpc.Core.RpcException? Fault { get; set; }
public Task DispatchAsync(
string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct)
{
lock (_gate) _calls.Add(new Dial(nodeEndpoint, envelope, clusterId));
if (BlockBeforeFirstAwait)
{
// Blocks the CALLING thread, exactly as IMeshPskProvider.TryGetPsk does on a cold
// cache: an async method runs synchronously to its first await, so a dial invoked
// straight from the actor would park the mailbox here.
_released.Wait(TimeSpan.FromSeconds(30));
}
var shouldFail =
(FailEndpointsContaining is not null && nodeEndpoint.Contains(FailEndpointsContaining))
|| FailClusters.Contains(clusterId, StringComparer.OrdinalIgnoreCase);
return shouldFail && Fault is not null
? Task.FromException(Fault)
: Task.CompletedTask;
}
/// <summary>One recorded dial.</summary>
public sealed record Dial(string Endpoint, CommandEnvelope Envelope, string ClusterId);
}
/// <summary>Records the gate and invalidation calls the contact refresh is supposed to make.</summary>
private sealed class RecordingPskProvider : IMeshPskProvider
{
private readonly object _gate = new();
private readonly List<IReadOnlyList<string>> _publications = [];
private int _invalidateAllCount;
/// <summary>Every known-cluster set published, in order.</summary>
public IReadOnlyList<IReadOnlyList<string>> KnownClusterPublications
{
get { lock (_gate) return _publications.ToList(); }
}
/// <summary>How many times the whole cache was dropped.</summary>
public int InvalidateAllCount => Volatile.Read(ref _invalidateAllCount);
public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk)
{
psk = "psk";
return true;
}
public bool Matches(string clusterId, string presented) => true;
public void SetKnownClusters(IReadOnlyCollection<string> clusterIds)
{
lock (_gate) _publications.Add(clusterIds.ToList());
}
public void Invalidate(string clusterId) { }
public void InvalidateAll() => Interlocked.Increment(ref _invalidateAllCount);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs
+271
View File
@@ -0,0 +1,271 @@
using System.Diagnostics.CodeAnalysis;
using Grpc.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication;
/// <summary>
/// ClusterClient→gRPC Phase 1A — central's outbound command dial. One node, one call, no retry.
/// </summary>
/// <remarks>
/// The facts that matter are the credential ones: the bearer must be the TARGET node's cluster's
/// key (not central's, and not some other cluster's), and a cluster with no resolvable key must
/// never be dialled at all. Everything else is plumbing.
/// </remarks>
public sealed class MeshGrpcCommandDialerTests
{
private const string Endpoint = "http://site-a-1:4056";
private static CommandEnvelope Envelope() => new()
{
CorrelationId = Guid.NewGuid().ToString("N"),
RestartDriver = new RestartDriverCommand
{
ClusterId = "SITE-A",
DriverInstanceId = "drv-1",
ActorByUserName = "operator",
CorrelationId = Guid.NewGuid().ToString("N"),
},
};
private static MeshGrpcCommandDialer CreateDialer(
IMeshPskProvider psk,
RecordingClient client,
int dispatchTimeoutSeconds = 10)
=> new(
psk,
Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = dispatchTimeoutSeconds }),
_ => client);
[Fact]
public async Task Dial_presents_the_target_clusters_key_and_names_that_cluster()
{
// THE assertion. Central holds N keys; sending the wrong one turns a legitimate command into
// a PermissionDenied, and sending a fleet-wide one would be the blast radius this design
// exists to avoid.
var client = new RecordingClient();
var dialer = CreateDialer(
new FakePskProvider(("SITE-A", "psk-site-a"), ("SITE-B", "psk-site-b")), client);
await dialer.DispatchAsync(Endpoint, Envelope(), "SITE-B", CancellationToken.None);
client.LastHeaders.ShouldNotBeNull();
client.LastHeaders!.GetValue("authorization").ShouldBe("Bearer psk-site-b");
client.LastHeaders.GetValue("x-otopcua-cluster").ShouldBe("SITE-B");
}
[Fact]
public async Task A_cluster_with_no_key_is_refused_locally_and_never_dialled()
{
// Fail-closed: not "dial and let the node reject it". An unauthenticated command attempt
// against a node's control surface should not exist even as a rejected one.
var client = new RecordingClient();
var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client);
var ex = await Should.ThrowAsync<RpcException>(
() => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-Z", CancellationToken.None));
ex.StatusCode.ShouldBe(StatusCode.Unauthenticated);
client.Calls.ShouldBe(0);
}
[Fact]
public async Task The_envelope_reaches_the_node_unchanged()
{
var client = new RecordingClient();
var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client);
var envelope = Envelope();
await dialer.DispatchAsync(Endpoint, envelope, "SITE-A", CancellationToken.None);
client.LastRequest.ShouldBe(envelope);
}
[Fact]
public async Task A_node_rejection_surfaces_as_its_own_RpcException()
{
// The actor grades per-node outcomes off the status code, so it must arrive intact rather
// than wrapped in an AggregateException or swallowed here.
var client = new RecordingClient
{
Fault = new RpcException(new Status(StatusCode.PermissionDenied, "nope")),
};
var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client);
var ex = await Should.ThrowAsync<RpcException>(
() => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None));
ex.StatusCode.ShouldBe(StatusCode.PermissionDenied);
}
[Fact]
public async Task The_per_call_deadline_bounds_a_node_that_never_answers()
{
// A dispatch dials every node in the fleet. Without this bound one unreachable node holds a
// task and a socket for as long as the process lives, while the deploy path's own apply
// deadline expires around it.
var client = new RecordingClient { HangUntilCancelled = true };
var dialer = CreateDialer(
new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 1);
await Should.ThrowAsync<OperationCanceledException>(
() => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None)
.WaitAsync(TimeSpan.FromSeconds(10)));
}
[Fact]
public async Task The_callers_token_still_cancels_a_dial_inside_the_deadline()
{
// Linked, not replaced: actor shutdown must cancel everything in flight even when the
// per-call deadline is nowhere near expiring.
var client = new RecordingClient { HangUntilCancelled = true };
var dialer = CreateDialer(
new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 600);
using var cts = new CancellationTokenSource();
var dial = dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", cts.Token);
await cts.CancelAsync();
await Should.ThrowAsync<OperationCanceledException>(() => dial.WaitAsync(TimeSpan.FromSeconds(10)));
}
[Fact]
public async Task A_zero_dispatch_timeout_leaves_the_dial_bounded_only_by_the_callers_token()
{
// The validator requires > 0 under Mode=Grpc, so this is the misconfigured/other-mode shape.
// It must not be read as "expire immediately" — that would drop every command instantly.
var client = new RecordingClient();
var dialer = CreateDialer(
new FakePskProvider(("SITE-A", "psk-site-a")), client, dispatchTimeoutSeconds: 0);
await dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None);
client.Calls.ShouldBe(1);
}
[Fact]
public async Task Dialing_after_disposal_throws_rather_than_repopulating_the_channel_cache()
{
var client = new RecordingClient();
var dialer = CreateDialer(new FakePskProvider(("SITE-A", "psk-site-a")), client);
dialer.Dispose();
await Should.ThrowAsync<ObjectDisposedException>(
() => dialer.DispatchAsync(Endpoint, Envelope(), "SITE-A", CancellationToken.None));
}
[Fact]
public async Task The_real_channel_factory_dials_and_is_disposable()
{
// The fake client seam covers every other fact here, which means nothing else ever exercises
// GrpcChannel.ForAddress or the cache it fills. Dial an endpoint nothing is listening on:
// the failure must be an ordinary RpcException the actor can log, not a construction throw,
// and Dispose must then clean up the channel the dial created.
var dialer = new MeshGrpcCommandDialer(
new FakePskProvider(("SITE-A", "psk-site-a")),
Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 5 }));
await Should.ThrowAsync<RpcException>(
() => dialer.DispatchAsync(
"http://127.0.0.1:1", Envelope(), "SITE-A", CancellationToken.None));
Should.NotThrow(dialer.Dispose);
}
[Fact]
public void It_activates_from_the_container_despite_the_optional_client_factory_parameter()
{
// The seam is an optional constructor parameter, which MS.DI honours only because it checks
// HasDefaultValue. Get that wrong — by making it non-optional, or by adding a second public
// constructor — and the failure is a startup crash on the admin node, invisible to every
// other test here because they all construct the dialer by hand.
var services = new ServiceCollection();
services.AddSingleton<IMeshPskProvider>(new FakePskProvider(("SITE-A", "psk-site-a")));
services.AddSingleton<IOptions<MeshTransportOptions>>(
Options.Create(new MeshTransportOptions { DispatchTimeoutSeconds = 10 }));
services.AddSingleton<IMeshCommandDialer, MeshGrpcCommandDialer>();
using var provider = services.BuildServiceProvider();
provider.GetRequiredService<IMeshCommandDialer>().ShouldBeOfType<MeshGrpcCommandDialer>();
}
/// <summary>
/// A <see cref="NodeCommandService.NodeCommandServiceClient"/> that records what it was
/// handed and answers however the test asks. The generated client's protected parameterless
/// constructor exists precisely for this.
/// </summary>
private sealed class RecordingClient : NodeCommandService.NodeCommandServiceClient
{
/// <summary>Number of calls that reached the client.</summary>
public int Calls { get; private set; }
/// <summary>Headers of the most recent call.</summary>
public Metadata? LastHeaders { get; private set; }
/// <summary>Request of the most recent call.</summary>
public CommandEnvelope? LastRequest { get; private set; }
/// <summary>When set, the call faults with this instead of answering.</summary>
public RpcException? Fault { get; set; }
/// <summary>When true, the call never completes until its token is cancelled.</summary>
public bool HangUntilCancelled { get; set; }
public override AsyncUnaryCall<ExecuteAck> ExecuteAsync(
CommandEnvelope request,
Metadata? headers = null,
DateTime? deadline = null,
CancellationToken cancellationToken = default)
{
Calls++;
LastHeaders = headers;
LastRequest = request;
var response = Fault is not null
? Task.FromException<ExecuteAck>(Fault)
: HangUntilCancelled
? HangAsync(cancellationToken)
: Task.FromResult(new ExecuteAck());
return new AsyncUnaryCall<ExecuteAck>(
response,
Task.FromResult(new Metadata()),
() => Status.DefaultSuccess,
() => [],
() => { });
}
private static async Task<ExecuteAck> HangAsync(CancellationToken ct)
{
await Task.Delay(Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false);
return new ExecuteAck();
}
}
/// <summary>An <see cref="IMeshPskProvider"/> over a fixed key set; unknown clusters fail closed.</summary>
private sealed class FakePskProvider(params (string Cluster, string Psk)[] keys) : IMeshPskProvider
{
private readonly Dictionary<string, string> _keys =
keys.ToDictionary(k => k.Cluster, k => k.Psk, StringComparer.OrdinalIgnoreCase);
public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk)
=> _keys.TryGetValue(clusterId, out psk);
public bool Matches(string clusterId, string presented)
=> _keys.TryGetValue(clusterId, out var expected)
&& string.Equals(expected, presented, StringComparison.Ordinal);
public void SetKnownClusters(IReadOnlyCollection<string> clusterIds) { }
public void Invalidate(string clusterId) { }
public void InvalidateAll() { }
}
}