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

feat(mesh): CentralCommandService on the ConfigServe listener

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-06 05:21:17 -04:00
parent a400b41502
commit 7e7df337cc
5 changed files with 518 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs
+108 -2
View File
@@ -20,9 +20,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;
[Collection(RoleEnvironmentCollection.Name)]
public class MeshTransportOptionsValidatorTests
{
private static ValidateOptionsResult Validate(MeshTransportOptions o, params string[] roles)
private static ValidateOptionsResult Validate(MeshTransportOptions o, params string[] roles) =>
Validate(o, new Dictionary<string, string?>(), roles);
/// <summary>
/// As <see cref="Validate(MeshTransportOptions, string[])"/>, plus arbitrary configuration
/// keys — the admin-side rule cross-reads <c>ConfigServe:GrpcListenPort</c>, which lives
/// outside this options class entirely.
/// </summary>
private static ValidateOptionsResult Validate(
MeshTransportOptions o, Dictionary<string, string?> extra, params string[] roles)
{
var pairs = new Dictionary<string, string?>();
var pairs = new Dictionary<string, string?>(extra);
for (var i = 0; i < roles.Length; i++)
{
pairs[$"Cluster:Roles:{i}"] = roles[i];
@@ -34,6 +43,9 @@ public class MeshTransportOptionsValidatorTests
.Validate(MeshTransportOptions.SectionName, o);
}
private static Dictionary<string, string?> ConfigServePort(string? value) =>
new() { ["ConfigServe:GrpcListenPort"] = value };
private static MeshTransportOptions WellFormedGrpc() => new()
{
Mode = MeshTransportOptions.ModeGrpc,
@@ -298,11 +310,105 @@ public class MeshTransportOptionsValidatorTests
// The node-side surface is what a DRIVER uses to dial central and post its ack. An admin-only
// central node dials nodes instead and resolves per-cluster keys at runtime, so requiring
// these here would make an admin node carry a PSK and an endpoint list it never reads.
// (The ConfigServe port IS required of an admin node under Grpc — supplied here so this test
// stays about the node-side keys and nothing else.)
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("4055"),
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void Grpc_admin_node_without_a_config_serve_listener_fails()
{
// Central hosts CentralCommandService on the ConfigServe h2c listener, so an unset port means
// the service is never mapped and every node's ack has nowhere to land. Nothing errors at
// runtime — the deployment simply times out naming nodes that applied perfectly, which is the
// exact absence this validator exists to convert into a refusal to start.
var result = Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
"admin", "cluster-MAIN");
result.Failed.ShouldBeTrue();
result.FailureMessage.ShouldContain("ConfigServe:GrpcListenPort");
result.FailureMessage.ShouldContain("ConfigServe__GrpcListenPort");
}
[Fact]
public void Grpc_admin_node_with_a_zero_config_serve_port_fails()
{
// 0 is the documented "disabled" value, not merely a missing key — it must fail exactly as an
// absent one does, because it binds nothing just as surely.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("0"),
"admin", "cluster-MAIN").Failed.ShouldBeTrue();
}
[Fact]
public void Grpc_admin_node_with_an_unparseable_config_serve_port_fails()
{
// A typo'd port binds nothing, so it belongs on the same side of the line as an absent one.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("40o55"),
"admin", "cluster-MAIN").Failed.ShouldBeTrue();
}
[Fact]
public void Grpc_admin_node_with_a_config_serve_listener_passes()
{
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc },
ConfigServePort("4055"),
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void Grpc_driver_only_node_needs_no_config_serve_listener()
{
// A driver node dials central; it serves no central command endpoint at all, and requiring an
// admin-side listener port of it would demand configuration it never reads.
Validate(WellFormedGrpc(), "driver", "cluster-SITE-A").Succeeded.ShouldBeTrue();
}
[Fact]
public void Dps_admin_node_needs_no_config_serve_listener()
{
// The no-new-obligations invariant, admin side: a central node still on the default transport
// publishes on DistributedPubSub and hosts no command endpoint.
Validate(
new MeshTransportOptions { Mode = MeshTransportOptions.ModeDps },
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void ClusterClient_admin_node_needs_no_config_serve_listener()
{
// Contact points are required by the mode itself, independently of role — supplied so this
// test fails only if the ConfigServe rule wrongly fires outside Grpc mode.
Validate(
new MeshTransportOptions
{
Mode = MeshTransportOptions.ModeClusterClient,
CentralContactPoints = ["akka.tcp://otopcua@central-1:4053"],
},
"admin", "cluster-MAIN").Succeeded.ShouldBeTrue();
}
[Fact]
public void The_admin_role_is_also_read_from_OTOPCUA_ROLES()
{
// Same roles-source divergence as the driver case (Gitea #532), on the side that hosts the
// endpoint: a central node configured only through the env var is still the node whose
// listener every ack depends on.
using (new EnvVarScope("OTOPCUA_ROLES", "admin,cluster-MAIN"))
{
Validate(new MeshTransportOptions { Mode = MeshTransportOptions.ModeGrpc })
.Failed.ShouldBeTrue();
}
}
[Fact]
public void Grpc_role_less_node_needs_none_of_the_node_side_keys()
{
tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs
+246
View File
@@ -0,0 +1,246 @@
using Akka.Hosting;
using Akka.TestKit.Xunit2;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.ControlPlane;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Grpc;
/// <summary>
/// The central-side gRPC ingress for a node's deployment ack (ClusterClient→gRPC migration
/// Phase 1B). It decodes the wire message and hands it to the registered
/// <c>CentralCommunicationActor</c> — the same actor the ClusterClient receptionist delivered to,
/// whose forward-to-the-coordinator path is untouched.
/// </summary>
/// <remarks>
/// Both failure modes matter more than the happy path. An ack that cannot be decoded must be
/// refused rather than guessed at (a wrong outcome seals a deployment nobody applied), and an ack
/// arriving before the actor exists must say <see cref="StatusCode.Unavailable"/> so the caller
/// retries instead of recording a permanent failure against a node that is fine.
/// </remarks>
public sealed class CentralCommandGrpcServiceTests : TestKit
{
private static ApplyAckMessage WellFormed() => new()
{
DeploymentId = Guid.NewGuid().ToString("N"),
NodeId = "site-a-1",
Outcome = ApplyOutcome.Applied,
CorrelationId = Guid.NewGuid().ToString("N"),
};
private static CentralCommandGrpcService Service(ActorRegistry registry) =>
new(registry, NullLogger<CentralCommandGrpcService>.Instance);
[Fact]
public async Task A_well_formed_ack_is_decoded_and_told_to_the_registered_actor()
{
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var request = WellFormed();
request.FailureReason = "disk full";
request.Outcome = ApplyOutcome.Failed;
var response = await Service(registry).ReportApply(request, new FakeServerCallContext());
response.ShouldNotBeNull();
var received = probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3));
received.DeploymentId.ShouldBe(DeploymentId.Parse(request.DeploymentId));
received.NodeId.Value.ShouldBe("site-a-1");
received.Outcome.ShouldBe(ApplyAckOutcome.Failed);
received.FailureReason.ShouldBe("disk full");
received.CorrelationId.ToString().ShouldBe(request.CorrelationId);
}
[Fact]
public async Task An_absent_failure_reason_arrives_as_null_and_an_empty_one_as_empty()
{
// Presence, not emptiness — the mapper's contract, asserted at the seam that actually feeds
// the coordinator. Collapsing the two would report an operator-visible blank reason as "no
// reason recorded".
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var service = Service(registry);
await service.ReportApply(WellFormed(), new FakeServerCallContext());
probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3)).FailureReason.ShouldBeNull();
var blank = WellFormed();
blank.FailureReason = "";
await service.ReportApply(blank, new FakeServerCallContext());
probe.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(3)).FailureReason.ShouldBe("");
}
[Fact]
public async Task An_unregistered_actor_yields_Unavailable()
{
// The boot window: the listener answers before Akka.Hosting has spawned the actor. Unavailable
// is the retryable status, so the node's next attempt lands instead of the deployment
// recording a permanent failure against a node that applied fine.
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(new ActorRegistry()).ReportApply(WellFormed(), new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.Unavailable);
}
[Fact]
public async Task An_unspecified_outcome_yields_InvalidArgument()
{
// proto3 decodes an absent enum as 0. Refusing is the whole point: the deploy path seals on
// Applied, so guessing an outcome the node never sent would seal a deployment nobody applied.
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var request = WellFormed();
request.Outcome = ApplyOutcome.Unspecified;
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task A_malformed_id_yields_InvalidArgument()
{
var request = WellFormed();
request.DeploymentId = "not-a-guid";
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task A_blank_node_id_yields_InvalidArgument()
{
// NodeId.Parse refuses a blank name; an ack whose node cannot be named is uncountable at the
// coordinator, so it must be refused here rather than forwarded as an anonymous ack.
var request = WellFormed();
request.NodeId = "";
var probe = CreateTestProbe("central-comm");
var registry = new ActorRegistry();
registry.Register<CentralCommunicationActorKey>(probe.Ref);
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(registry).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
[Fact]
public async Task An_undecodable_message_is_graded_before_the_registry()
{
// Ordering pin. With NO actor registered AND an undecodable message, the answer is
// InvalidArgument, not Unavailable: Unavailable invites a retry, and retrying a message that
// can never decode is a loop. The caller's fault is reported as the caller's fault.
var request = WellFormed();
request.Outcome = ApplyOutcome.Unspecified;
var ex = await Should.ThrowAsync<RpcException>(() =>
Service(new ActorRegistry()).ReportApply(request, new FakeServerCallContext()));
ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
}
[Fact]
public async Task A_decode_failure_logs_the_raw_correlation_id()
{
// The correlation id is logged as the raw wire string precisely because the ids may be what
// failed to parse — an operator correlating a rejection cannot be given a value that only
// exists once decoding succeeded.
var logger = new CapturingLogger<CentralCommandGrpcService>();
var request = WellFormed();
request.CorrelationId = "correlation-that-never-parses";
request.DeploymentId = "not-a-guid";
await Should.ThrowAsync<RpcException>(() =>
new CentralCommandGrpcService(new ActorRegistry(), logger)
.ReportApply(request, new FakeServerCallContext()));
logger.Entries.Any(e => e.Message.Contains("correlation-that-never-parses", StringComparison.Ordinal))
.ShouldBeTrue("the rejection log must carry the raw correlation id");
}
/// <summary>Test logger that captures all log entries.</summary>
/// <typeparam name="T">The category type.</typeparam>
private sealed class CapturingLogger<T> : ILogger<T>
{
/// <summary>Gets the captured log entries.</summary>
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = [];
/// <summary>Begins a scope (stub implementation).</summary>
/// <typeparam name="TState">The type of the scope state.</typeparam>
/// <param name="state">The scope state.</param>
/// <returns>A no-op scope.</returns>
public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
/// <summary>Checks if logging is enabled (always true).</summary>
/// <param name="logLevel">The log level to check.</param>
/// <returns><c>true</c>.</returns>
public bool IsEnabled(LogLevel logLevel) => true;
/// <summary>Logs an entry and captures it.</summary>
/// <typeparam name="TState">The type of the log state.</typeparam>
/// <param name="logLevel">The log level.</param>
/// <param name="eventId">The event ID.</param>
/// <param name="state">The log state.</param>
/// <param name="exception">The exception, if any.</param>
/// <param name="formatter">The log message formatter.</param>
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception), exception));
}
private sealed class NullScope : IDisposable
{
/// <summary>Gets the singleton instance.</summary>
public static NullScope Instance { get; } = new();
/// <summary>No-op.</summary>
public void Dispose() { }
}
}
/// <summary>Minimal <see cref="ServerCallContext"/> — the service reads only the method path.</summary>
private sealed class FakeServerCallContext : ServerCallContext
{
protected override string MethodCore => "/mesh_transport.v1.CentralCommandService/ReportApply";
protected override string HostCore => "localhost";
protected override string PeerCore => "ipv4:127.0.0.1:0";
protected override DateTime DeadlineCore => DateTime.MaxValue;
protected override Metadata RequestHeadersCore => [];
protected override CancellationToken CancellationTokenCore => CancellationToken.None;
protected override Metadata ResponseTrailersCore { get; } = [];
protected override Status StatusCore { get; set; }
protected override WriteOptions? WriteOptionsCore { get; set; }
protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
throw new NotSupportedException();
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
}
}