diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs new file mode 100644 index 00000000..473c4366 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs @@ -0,0 +1,110 @@ +using Akka.Actor; +using Akka.Hosting; +using Grpc.Core; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Runtime; +// The generated service container is itself named NodeCommandService; alias its nested server base so +// this impl can keep the natural name without colliding with the generated type. +using GeneratedServiceBase = + ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1.NodeCommandService.NodeCommandServiceBase; +// Akka.Actor (imported for IActorRef.Tell) declares a Status of its own; only the gRPC one is used here. +using Status = Grpc.Core.Status; + +namespace ZB.MOM.WW.OtOpcUa.Host.Grpc; + +/// +/// Node-side command ingress (ClusterClient→gRPC migration Phase 1). Central dials a driver node +/// and calls ; the envelope is decoded back into the same CLR record the +/// ClusterClient receptionist used to deliver and handed to this node's +/// NodeCommunicationActor, which re-emits it on the node's EventStream exactly as +/// before. Nothing downstream of that actor can tell which transport carried the command. +/// +/// +/// +/// One-way, verbatim from the transport it replaces. ExecuteAck means "decoded and +/// handed to the comm actor" — not applied, and not acknowledged by any EventStream subscriber. +/// The Tell is deliberately not an Ask: no migrated command has a reply, and +/// waiting for one would invent a semantic central does not have. +/// +/// +/// Decode runs BEFORE the registry lookup, and the two failures are graded differently. +/// An undecodable envelope is — permanent, and it will +/// never decode on a retry. A call arriving before AddOtOpcUaRuntime has spawned and +/// registered the comm actor is — the house readiness +/// convention, transient, and correct to retry. Checking readiness first would report a +/// permanently malformed command as a readiness problem and invite an unbounded retry loop. +/// +/// +/// Every log line carries the envelope's own correlation_id, read as a raw string. +/// Three of the four commands also carry a domain correlation, but AlarmCommand has no +/// CLR member for one, so the envelope field is the only id every command shares — and it is +/// still readable when the inner fields are what failed to parse, which is precisely when an +/// operator needs it. +/// +/// +/// Inbound auth is MeshCommandAuthInterceptor, scoped to this service's method prefix and +/// registered at the AddGrpc level; it fail-closes with no MeshTransport:GrpcPsk +/// configured. Mapping this service is what makes that gate live. +/// +/// +public sealed class NodeCommandGrpcService : GeneratedServiceBase +{ + private readonly ActorRegistry _registry; + private readonly ILogger _log; + + /// Initializes a new instance of the class. + /// Actor registry holding this node's NodeCommunicationActor. + /// Logger. + public NodeCommandGrpcService(ActorRegistry registry, ILogger log) + { + _registry = registry; + _log = log; + } + + /// + public override Task Execute(CommandEnvelope request, ServerCallContext context) + { + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(context); + + object command; + try + { + command = MeshTransportDtoMapper.FromEnvelope(request); + } + catch (Exception ex) when (ex is InvalidOperationException or FormatException or ArgumentException) + { + // The envelope carries no credential material, so logging its correlation id is safe and is + // the only handle an operator has on a call whose inner fields did not parse. + _log.LogWarning( + ex, + "Rejected a mesh command (correlationId={CorrelationId}, oneof={CommandCase}): the " + + "envelope could not be decoded. The command was NOT delivered.", + request.CorrelationId, request.CommandCase); + throw new RpcException(new Status( + StatusCode.InvalidArgument, "Command envelope could not be decoded.")); + } + + if (!_registry.TryGet(out var comm)) + { + _log.LogWarning( + "Rejected a mesh command (correlationId={CorrelationId}, type={CommandType}): this node's " + + "communication actor is not registered yet. The node is still starting; the command was " + + "NOT delivered and central may retry.", + request.CorrelationId, command.GetType().Name); + throw new RpcException(new Status( + StatusCode.Unavailable, "Node is still starting: the command endpoint is not ready.")); + } + + comm.Tell(command); + + _log.LogDebug( + "Mesh: accepted {CommandType} (correlationId={CorrelationId}) over gRPC; handed to the node " + + "communication actor.", + command.GetType().Name, request.CorrelationId); + + return Task.FromResult(new ExecuteAck()); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 8fb24e89..33ff5a43 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -740,9 +740,18 @@ if (hasAdmin && configServeGrpcPort > 0) // Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by // TelemetryStreamAuthInterceptor (fail-closed on empty Telemetry:ApiKey). +// +// The node-side command endpoint (ClusterClient→gRPC migration Phase 1) rides the SAME driver-role +// h2c listener rather than claiming a fourth port: central already dials this node here, both +// directions are per-cluster-keyed, and a separate port would be one more thing to open on every +// site firewall for no isolation gain. Mapping it is what makes MeshCommandAuthInterceptor live — +// it is registered above and is a pass-through until a call matches this service's method prefix. +// It fail-closes with no MeshTransport:GrpcPsk, so an operator who maps the service without setting +// a key gets a closed endpoint, not an open one. if (hasDriver && telemetryListenPort > 0) { app.MapGrpcService(); + app.MapGrpcService(); } app.MapOtOpcUaHealth(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs index 63dc64f4..26a08294 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs @@ -2,5 +2,6 @@ using Xunit; // The TelemetryStreamGrpcService concurrency counter is process-wide (static). Run this assembly's // tests serially so the cap test observes a deterministic active-stream count with no cross-test -// interference. This project holds only telemetry tests, so serial execution costs nothing. +// interference. Every other suite here is small and fast (the mesh-command tests spin a TestKit +// ActorSystem per class), so serial execution costs nothing. [assembly: CollectionBehavior(DisableTestParallelization = true)] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs new file mode 100644 index 00000000..0ccd6ea8 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs @@ -0,0 +1,186 @@ +using Akka.Hosting; +using Akka.TestKit.Xunit2; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +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.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Host.Grpc; +using ZB.MOM.WW.OtOpcUa.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Grpc; + +/// +/// Verifies the node-side command ingress (ClusterClient→gRPC migration Phase 1): each of the four +/// central→node commands is decoded off its and handed to the +/// registered NodeCommunicationActor as the same CLR record the ClusterClient transport +/// delivered, an undecodable envelope is refused with , and +/// a call landing in the boot window before the actor is registered is refused with +/// . +/// +/// +/// The actor is a TestProbe registered under : what +/// this service owes the rest of the system is the exact message the real actor already knows how to +/// re-emit, so asserting on what arrives at the registry key is the whole contract. The onward leg — +/// comm actor to EventStream to the real driver host — is pinned unchanged by +/// EventStreamCommandDeliveryTests in the Runtime test project. +/// +public sealed class NodeCommandGrpcServiceTests : TestKit +{ + private static readonly CorrelationId AlarmCorrelation = new(Guid.NewGuid()); + + private static ServerCallContext Ctx() => new FakeServerCallContext(); + + private static NodeCommandGrpcService NewService(ActorRegistry registry) => + new(registry, NullLogger.Instance); + + [Fact] + public async Task Each_of_the_four_commands_reaches_the_comm_actor_as_its_domain_record() + { + var probe = CreateTestProbe(); + var registry = new ActorRegistry(); + registry.Register(probe.Ref); + var service = NewService(registry); + + var dispatch = new DispatchDeployment( + new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid())); + var restart = new RestartDriver("cluster-a", "drv-1", "alice", Guid.NewGuid()); + var reconnect = new ReconnectDriver("cluster-a", "drv-2", "bob", Guid.NewGuid()); + var alarm = new AlarmCommand("alarm-7", "Acknowledge", "carol", "looked at it", null); + + (await service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx())).ShouldNotBeNull(); + probe.ExpectMsg().ShouldBe(dispatch); + + await service.Execute(MeshTransportDtoMapper.ToEnvelope(restart), Ctx()); + probe.ExpectMsg().ShouldBe(restart); + + await service.Execute(MeshTransportDtoMapper.ToEnvelope(reconnect), Ctx()); + probe.ExpectMsg().ShouldBe(reconnect); + + await service.Execute(MeshTransportDtoMapper.ToEnvelope(alarm, AlarmCorrelation), Ctx()); + probe.ExpectMsg().ShouldBe(alarm); + } + + [Fact] + public async Task A_call_in_the_boot_window_before_the_actor_is_registered_is_unavailable() + { + // Nothing registered: the ActorSystem exists but AddOtOpcUaRuntime has not spawned the comm + // actor yet. Unavailable (not Internal) because the condition is transient and retryable — + // central's dialer may try the same node again a moment later and succeed. + var service = NewService(new ActorRegistry()); + var dispatch = new DispatchDeployment( + new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid())); + + var ex = await Should.ThrowAsync( + () => service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx())); + + ex.StatusCode.ShouldBe(StatusCode.Unavailable); + } + + [Fact] + public async Task An_unset_oneof_is_invalid_argument() + { + var probe = CreateTestProbe(); + var registry = new ActorRegistry(); + registry.Register(probe.Ref); + var service = NewService(registry); + + var ex = await Should.ThrowAsync(() => service.Execute( + new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx())); + + ex.StatusCode.ShouldBe(StatusCode.InvalidArgument); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + [Theory] + [MemberData(nameof(UndecodableEnvelopes))] + public async Task An_envelope_whose_fields_do_not_parse_is_invalid_argument(CommandEnvelope envelope) + { + var probe = CreateTestProbe(); + var registry = new ActorRegistry(); + registry.Register(probe.Ref); + var service = NewService(registry); + + var ex = await Should.ThrowAsync(() => service.Execute(envelope, Ctx())); + + ex.StatusCode.ShouldBe(StatusCode.InvalidArgument); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + public static TheoryData UndecodableEnvelopes() => + [ + // Deployment id is not a hyphenless GUID -> FormatException out of the mapper. + new CommandEnvelope + { + CorrelationId = Guid.NewGuid().ToString("N"), + DispatchDeployment = new DispatchDeploymentCommand + { + DeploymentId = "not-a-guid", + RevisionHash = "rev-1", + CorrelationId = Guid.NewGuid().ToString("N"), + }, + }, + // Revision hash is blank -> ArgumentException out of RevisionHash.Parse. + new CommandEnvelope + { + CorrelationId = Guid.NewGuid().ToString("N"), + DispatchDeployment = new DispatchDeploymentCommand + { + DeploymentId = Guid.NewGuid().ToString("N"), + RevisionHash = "", + CorrelationId = Guid.NewGuid().ToString("N"), + }, + }, + // Restart command with an unparseable correlation id -> FormatException. + new CommandEnvelope + { + CorrelationId = Guid.NewGuid().ToString("N"), + RestartDriver = new RestartDriverCommand + { + ClusterId = "cluster-a", + DriverInstanceId = "drv-1", + ActorByUserName = "alice", + CorrelationId = "nope", + }, + }, + ]; + + [Fact] + public async Task An_undecodable_envelope_in_the_boot_window_is_invalid_argument_not_unavailable() + { + // Ordering pin: decode runs BEFORE the registry lookup. Unavailable invites the caller to + // retry, and an envelope this build cannot decode will never decode — reporting it as a + // readiness problem would turn a permanent fault into an unbounded retry loop. + var service = NewService(new ActorRegistry()); + + var ex = await Should.ThrowAsync(() => service.Execute( + new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx())); + + ex.StatusCode.ShouldBe(StatusCode.InvalidArgument); + } + + /// Minimal ; this RPC reads nothing off the context but its method name. + private sealed class FakeServerCallContext : ServerCallContext + { + protected override string MethodCore => "/mesh_transport.v1.NodeCommandService/Execute"; + 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>()); + + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) => + throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj index 7798ef15..b4d6f70e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj @@ -10,6 +10,9 @@ + +