diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshCommand.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshCommand.cs index a12ab795..010b7c76 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshCommand.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshCommand.cs @@ -6,7 +6,7 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; /// /// The legacy DPS topic. Carried so the router can publish under /// MeshTransport:Mode = Dps without a second dispatch table; ignored entirely under -/// ClusterClient mode, where the destination is an actor path rather than a topic. +/// Grpc mode, where the destination is a node endpoint rather than a topic. /// /// /// The command itself, never wrapped. Node-side handlers see exactly the type they see today, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs index 308117f1..99337b79 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshPaths.cs @@ -9,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; /// Declared once, here, rather than as literals on each side. The deployments / /// deployment-acks topic names are declared twice — once in /// ConfigPublishCoordinator and again in DriverHostActor — and agree only by -/// convention; renaming one would compile, deploy, and deliver nothing. A ClusterClient path +/// convention; renaming one would compile, deploy, and deliver nothing. A mesh path /// has the same failure mode and no compiler to catch it, so the constant is shared. /// /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs index 1b83ef7d..f3d78255 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs @@ -9,13 +9,13 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; /// /// The one translation between the five cross-mesh command records and their -/// mesh_transport.v1 wire shapes (ClusterClient→gRPC migration Phase 0). Both directions live +/// mesh_transport.v1 wire shapes (mesh gRPC command plane, Phase 0). Both directions live /// here so a field can only be added to one side without the other by making this file /// visibly asymmetric. /// /// /// -/// Nothing here guesses. The transport it replaces — an Akka ClusterClient receptionist — +/// Nothing here guesses. The transport it replaced — an Akka ClusterClient receptionist — /// dropped an undeliverable command silently, and the entire point of the migration is to stop /// doing that. So every decode that cannot produce a faithful record throws: an unset oneof, an /// APPLY_OUTCOME_UNSPECIFIED ack, an enum value from a newer peer. A loud failure on one diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto index d91c52f8..1dd09580 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto @@ -6,9 +6,9 @@ option csharp_namespace = "ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1"; import "google/protobuf/timestamp.proto"; -// ClusterClient→gRPC migration Phase 0: the cross-mesh command plane. The five messages that cross -// an application-Cluster boundary today ride an Akka ClusterClient receptionist; this contract is -// the replacement carrier. Four are central→node commands, one is the node→central deployment ack. +// The cross-mesh command plane (introduced in Phase 0, now the only transport that crosses an +// application-Cluster boundary). Five messages ride it: four central→node commands and one +// node→central deployment ack. // // The C# namespace deliberately says "MeshTransport", never "MeshCommand" — the latter is already // the CLR envelope record ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh.MeshCommand, and a generated type @@ -20,17 +20,17 @@ import "google/protobuf/timestamp.proto"; // MeshTransportProtoContract + its test in the Commons test project. // Central → node. Delivered to the node's NodeCommunicationActor, which re-emits the decoded message -// on the node's EventStream. ONE-WAY semantics, preserved verbatim from the ClusterClient transport: -// the ack means "received and re-emitted", nothing more — not applied, not acknowledged by any +// on the node's EventStream. ONE-WAY semantics, matching the node-local DistributedPubSub leg: the +// ack means "received and re-emitted", nothing more — not applied, not acknowledged by any // subscriber. No retry is layered on a failure; a failed Execute is logged per node and dropped, -// which is an honest upgrade over the receptionist's silent drop rather than a semantic change. +// which is an honest upgrade over a pub/sub publish nobody was listening for. service NodeCommandService { rpc Execute (CommandEnvelope) returns (ExecuteAck); } // Node → central. Delivered to central's CentralCommunicationActor, which forwards to the // ConfigPublishCoordinator singleton proxy — so an ack landing on either node of the central pair -// routes identically, exactly as the receptionist made arrival-node-agnostic. ONE-WAY semantics: the +// routes identically — which central node answered is deliberately not observable. ONE-WAY semantics: the // ack means "handed to the coordinator", not "the deployment is complete". service CentralCommandService { rpc ReportApply (ApplyAckMessage) returns (ReportAck); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/ClusterNode.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/ClusterNode.cs index 6214d08f..20653c20 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/ClusterNode.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Entities/ClusterNode.cs @@ -22,7 +22,7 @@ public sealed class ClusterNode /// Akka remoting port (default 4053). This is central's dial target, not the node's own /// binding configuration — the node binds from Cluster:Port in its own appsettings, /// and this column duplicates that value for a reader that cannot see the node's config. - /// Per-cluster mesh Phase 2 builds its ClusterClient contact points from + /// Per-cluster mesh Phase 2 built its Akka contact points from /// + this port, at which point central and the site node no longer share a /// gossip ring and central has no other way to learn it. /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs index 5fe5ea78..d70834a4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs @@ -17,10 +17,10 @@ using Status = Grpc.Core.Status; namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc; /// -/// Central-side ingress for a node's deployment acknowledgment (ClusterClient→gRPC migration +/// Central-side ingress for a node's deployment acknowledgment (mesh gRPC command plane /// Phase 1B). Decodes an and hands it to the node-local -/// CentralCommunicationActor — the same actor an Akka ClusterClient receptionist delivered -/// to, whose forward-to-the-coordinator path is untouched. Registered only on admin-role nodes +/// CentralCommunicationActor — the same actor the node-local DistributedPubSub path +/// delivers to, whose forward-to-the-coordinator path is untouched. Registered only on admin-role nodes /// and only when ConfigServe:GrpcListenPort > 0. /// /// @@ -52,7 +52,7 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc; /// /// One-way semantics, preserved. Tell, not Ask: the returned /// means "handed to the comm actor", never "the deployment is -/// complete" — the same promise the receptionist made. The comm actor forwards to the +/// complete" — the same promise the DistributedPubSub leg makes. The comm actor forwards to the /// coordinator singleton proxy, so an ack landing on either central node routes identically. /// /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs index 9a7de9d6..6f21f0e3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs @@ -92,8 +92,7 @@ public sealed class AdminOperationsActor : ReceiveActor /// retained, not scheduled for deletion — an earlier note here promised Phase 6 would /// remove it, which was deliberately reversed (docs/Configuration.md § "The compiled /// transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and - /// default transport for a single-mesh topology. What is being replaced is the ClusterClient - /// branch, by gRPC. + /// default transport for a single-mesh topology; gRPC is what crosses a mesh boundary. /// /// The legacy DistributedPubSub topic. /// The command. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs index 10d9761f..50b30fa5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/MeshGrpcCommandDialer.cs @@ -10,7 +10,7 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; /// /// Places one central→node command call on one node's NodeCommandService -/// (ClusterClient→gRPC migration Phase 1A). The seam CentralCommunicationActor fans +/// (mesh gRPC command plane, Phase 1A). The seam CentralCommunicationActor fans /// across its target set, kept as an interface so the actor's tests never open a socket. /// /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index f71c4c74..b52beb87 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -72,7 +72,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// not scheduled for deletion — an earlier note here promised Phase 6 would remove it, which /// was deliberately reversed (docs/Configuration.md § "The compiled transport-mode /// defaults were deliberately NOT flipped"): DPS is still the correct and default transport - /// for a single-mesh topology. What is being replaced is the ClusterClient branch, by gRPC. + /// for a single-mesh topology; gRPC is what crosses a mesh boundary. /// private readonly IActorRef? _meshRouter; private readonly ILoggingAdapter _log = Context.GetLogger(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj index 0ffdd314..05f6a470 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj @@ -19,7 +19,7 @@ Google.Protobuf + Grpc.Core.Api flow transitively from Commons (the generated client). --> - diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs index 3b1f635a..a201f1a3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/CentralCommandAuthInterceptor.cs @@ -6,7 +6,7 @@ using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// -/// Gates the central-side command endpoint (ClusterClient→gRPC migration). The +/// Gates the central-side command endpoint (mesh gRPC command plane). The /// CentralCommandService.ReportApply RPC is how a node tells central its deployment /// applied; the coordinator counts those acks, so anything able to reach central's command port /// could otherwise forge a fleet-wide "everything applied". This interceptor is the ONLY inbound diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs index 46a93954..273a3089 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/MeshCommandAuthInterceptor.cs @@ -9,7 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Cluster; namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// -/// Gates the node-side command endpoint (ClusterClient→gRPC migration). The +/// Gates the node-side command endpoint (mesh gRPC command plane). The /// NodeCommandService.Execute RPC is how central drives this node — restart a driver, /// acknowledge an alarm, dispatch a deployment — so anything able to reach a node's command port /// could otherwise operate the node. This interceptor is the ONLY inbound auth on that endpoint. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs index 473c4366..961a0357 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/NodeCommandGrpcService.cs @@ -15,11 +15,11 @@ 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 +/// Node-side command ingress (mesh gRPC command plane, 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 +/// node-local DistributedPubSub path delivers 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. +/// it would otherwise. Nothing downstream of that actor can tell which transport carried it. /// /// /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 80f3048b..1d4762e7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -527,20 +527,19 @@ if (hasAdmin) // hosted service has built the system). builder.Services.TryAddSingleton>(sp => () => sp.GetRequiredService()); builder.Services.AddSingleton(); - // Per-cluster mesh PSKs (ClusterClient→gRPC). Admin-only, and singleton because its whole job is + // Per-cluster mesh PSKs (mesh gRPC command plane). Admin-only, and singleton because its whole job is // to hold a cache: central talks to N clusters and would otherwise re-read the secret store on // every inbound ack. A node needs nothing of this — it holds exactly its own MeshTransport:GrpcPsk // — so registering it fleet-wide would give a driver-only node a lookup it can never use. // Depends on ISecretResolver, which AddOtOpcUaSecrets registers unconditionally below (DI // registration order is irrelevant; resolution order is not). builder.Services.AddSingleton(); - // Central's outbound command dial (ClusterClient→gRPC Phase 1A). Admin-only for the same reason + // Central's outbound command dial (mesh gRPC command plane, Phase 1A). Admin-only for the same reason // as the provider above — a node never dials another node — and singleton because it owns a // GrpcChannel per node endpoint; per-call channels would leak sockets across a fleet-wide fan. // Registered in every mode: CentralCommunicationActor only reaches it under Mode=Grpc, and an // idle registration costs nothing, whereas gating it here would mean flipping the mode on a - // running fleet needs a central restart before any command lands — the same reasoning that - // registers the ClusterClient receptionist unconditionally. + // running fleet needs a central restart before any command lands. builder.Services.AddSingleton(); } @@ -564,7 +563,7 @@ builder.Services.AddOtOpcUaObservability(builder.Configuration); // Registered whenever either role is present so an admin-only node still serves config and a // driver-only node still gates sync + telemetry. // -// The two mesh-command interceptors (ClusterClient→gRPC migration) join on the same terms: +// The two mesh-command interceptors (mesh gRPC command plane) join on the same terms: // MeshCommandAuthInterceptor gates the node-side NodeCommandService (driver-role, one key — this // node's own), CentralCommandAuthInterceptor gates the central-side CentralCommandService // (admin-role, one key per cluster via IMeshPskProvider). Both are registered NOW and are pure @@ -745,7 +744,7 @@ if (hasAdmin && configServeGrpcPort > 0) { app.MapGrpcService(); - // Central command ingress (ClusterClient→gRPC migration): the node→central deployment ack. It + // Central command ingress (mesh gRPC command plane): the node→central deployment ack. It // rides the SAME listener rather than claiming a port of its own — a second cleartext h2c port // would buy nothing but another firewall rule, and both endpoints are admin-role and dialled by // the same driver nodes. Deliberately NOT gated on MeshTransport:Mode: mapping it under Dps @@ -759,7 +758,7 @@ 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 +// The node-side command endpoint (mesh gRPC command plane, 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 — diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs index b47b2a96..56d9c7fe 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs @@ -12,14 +12,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// /// Posts this node's to central's CentralCommandService as a unary -/// gRPC call (ClusterClient→gRPC migration Task 1B.2), the MeshTransport:Mode = Grpc -/// replacement for the ClusterClient Send. +/// gRPC call (mesh gRPC command plane, Task 1B.2), the MeshTransport:Mode = Grpc +/// cross-mesh ack transport. /// /// /// /// One ordered pass, stopping at the first success. /// is an availability list, not a -/// fan-out list — the semantic of the ClusterClient.Send being replaced (Send, +/// fan-out list — the single-target semantic this leg owes (one ack per node per dispatch, /// deliberately not SendToAll). /// /// @@ -35,7 +35,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// depth and a bound on wasted work, not a correctness dependency. /// /// -/// No retry beyond that single pass, and no queue. The receptionist buffered nothing +/// No retry beyond that single pass, and no queue. The transport buffers nothing /// and retried nothing, and this is a like-for-like replacement, not an upgrade to delivery /// guarantees. The deploy path's own apply deadline stays the only recovery mechanism; a /// retry loop here would instead pile acks up behind an unreachable central and deliver a @@ -43,7 +43,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// /// /// All endpoints failing is a DROP — but a loud one. That is the same outcome the -/// ClusterClient path had when it could not reach central; what changes is that the operator +/// node has when it cannot reach central; what changes is that the operator /// now gets the node, the deployment, the correlation id and the status each endpoint /// answered with, which is what distinguishes "central is down" from "central rejected this /// node's key". diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs index 092d30ab..adeaf07c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs @@ -4,15 +4,15 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// /// Posts this node's deployment acknowledgment to central over the mesh_transport.v1 -/// command plane (ClusterClient→gRPC migration Phase 1B). The seam +/// command plane (mesh gRPC command plane, Phase 1B). The seam /// uses under MeshTransport:Mode = Grpc, in place of -/// the ClusterClient.Send it uses under ClusterClient. +/// the node-local DistributedPubSub publish it uses under Dps. /// /// /// /// One ack per node per dispatch. Whatever the implementation does about redundancy, it /// posts to one central endpoint at a time rather than fanning out — the semantic of the -/// ClusterClient.Send this replaces (Send, deliberately not SendToAll). +/// one-ack-per-node-per-dispatch semantic the deploy coordinator counts on. /// This is defence in depth, not a correctness dependency: the coordinator counts acks /// in a NodeId-keyed dictionary behind an expected-set membership guard /// (ConfigPublishCoordinator.HandleAck), so a duplicate from one node overwrites its own diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 7d14508e..4b865dab 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -374,9 +374,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient and = Grpc the node's - /// NodeCommunicationActor, which relays across the boundary over whichever transport it was - /// built with. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = Grpc the node's NodeCommunicationActor, which relays the ack + /// across the mesh boundary. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -459,9 +458,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient and = Grpc the node's - /// NodeCommunicationActor, which relays across the boundary over whichever transport it was - /// built with. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = Grpc the node's NodeCommunicationActor, which relays the ack + /// across the mesh boundary. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -584,11 +582,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _mediator.Tell( new Subscribe(ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RedundancyStateTopic, Self)); - // Per-cluster mesh Phase 2 dark switch. Under MeshTransport:Mode = ClusterClient these same - // commands arrive from NodeCommunicationActor on the node-local EventStream instead of the - // mesh-wide DPS topic. Subscribing to BOTH unconditionally is deliberate: exactly one of the - // two ever publishes, so the transport flag lives entirely on the publishing side and the - // switch can be flipped (or reverted) without touching any subscriber. + // Per-cluster mesh Phase 2. Under MeshTransport:Mode = Grpc these same commands arrive from + // NodeCommunicationActor on the node-local EventStream instead of the mesh-wide DPS topic. + // Subscribing to BOTH unconditionally is deliberate: exactly one of the two ever publishes, so + // the transport choice lives entirely on the publishing side and the mode can be flipped (or + // reverted) without touching any subscriber. Context.System.EventStream.Subscribe(Self, typeof(DispatchDeployment)); Context.System.EventStream.Subscribe(Self, typeof(RestartDriver)); Context.System.EventStream.Subscribe(Self, typeof(ReconnectDriver)); @@ -2492,11 +2490,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); if (_ackRouter is not null) { - // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = ClusterClient - // and = Grpc alike, the NodeCommunicationActor that relays this across the boundary (it - // owns the choice of transport, not this actor). Renamed from _coordinatorOverride in - // per-cluster mesh Phase 2: under those two modes this ref is emphatically NOT a - // coordinator, and the old name would send the next reader looking for one. + // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = Grpc, the + // NodeCommunicationActor that relays this across the mesh boundary (it owns the transport, + // not this actor). Renamed from _coordinatorOverride in per-cluster mesh Phase 2: under + // Grpc this ref is emphatically NOT a coordinator, and the old name would send the next + // reader looking for one. _ackRouter.Tell(ack); } else diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index 6d18058c..e4a9b208 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -627,7 +627,7 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor // Per-cluster mesh Phase 2 dark switch. An AdminUI-originated ack/shelve arrives from // NodeCommunicationActor on the node-local EventStream under - // MeshTransport:Mode = ClusterClient; the DPS subscribe above still carries the node-LOCAL + // MeshTransport:Mode = Grpc; the DPS subscribe above still carries the node-LOCAL // publisher (OtOpcUaServerHostedService, the OPC UA Part 9 method calls), which never // crosses the boundary and stays on DPS in both modes. Subscribing to both is safe: exactly // one publisher feeds each path. diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs index f93b6ebe..2291388f 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/MeshTransportOptionsValidatorTests.cs @@ -64,10 +64,10 @@ public class MeshTransportOptionsValidatorTests [Fact] public void Unknown_mode_fails() { - var result = Validate(new MeshTransportOptions { Mode = "receptionist" }); + var result = Validate(new MeshTransportOptions { Mode = "not-a-transport" }); result.Failed.ShouldBeTrue(); - result.FailureMessage.ShouldContain("receptionist"); + result.FailureMessage.ShouldContain("not-a-transport"); } [Fact] diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs index 47975f89..f11faf74 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/SplitTopologyTransportValidatorTests.cs @@ -94,7 +94,7 @@ public class SplitTopologyTransportValidatorTests // still a mode that delivers nothing across the boundary. The retired 'ClusterClient' value // lands here too. var result = Validate( - meshMode: "receptionist", + meshMode: "not-a-transport", telemetryMode: TelemetryOptions.ModeGrpc, telemetryDialMode: null, "driver", "cluster-SITE-A"); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs index 0177a85a..aa2adcee 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs @@ -12,7 +12,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Types; namespace ZB.MOM.WW.OtOpcUa.Commons.Tests; /// -/// Round-trip goldens for — the ClusterClient→gRPC command +/// Round-trip goldens for — the mesh gRPC command /// plane's only data contract. Every golden goes domain → proto → serialized bytes → proto → /// domain, because the failure class this suite exists to catch is not a mis-assigned property but a /// member that survives an in-memory copy and vanishes on the wire: proto3 explicit presence is a diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs index a7c33c08..fb5d7617 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportProtoContractTests.cs @@ -6,7 +6,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; namespace ZB.MOM.WW.OtOpcUa.Commons.Tests; /// -/// Contract-lock for the ClusterClient→gRPC command envelope's oneof. If a fifth command is added +/// Contract-lock for the mesh command envelope's oneof. If a fifth command is added /// to mesh_transport.proto without adding a matching entry to /// , this test goes red — the single guard /// that keeps the wire contract and the handled-case set from silently drifting apart. A drift diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ClusterNodeTransportPortsTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ClusterNodeTransportPortsTests.cs index 8b72f86d..0e7df9cd 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ClusterNodeTransportPortsTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/ClusterNodeTransportPortsTests.cs @@ -9,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Configuration.Tests; /// /// Round-trips / through /// the real schema (per-cluster mesh Phase 1). These are central's dial targets: Phase 2 -/// builds its ClusterClient contact points from Host + AkkaPort, and Phase 5 dials +/// built its Akka contact points from Host + AkkaPort, and Phase 5 dials /// GrpcPort for the telemetry stream. Central cannot see a site node's appsettings once /// the meshes split, so the value has to live in a row central can read. /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs index b2242985..5b3ccf99 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Grpc/CentralCommandGrpcServiceTests.cs @@ -14,9 +14,9 @@ using ZB.MOM.WW.OtOpcUa.ControlPlane; namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Grpc; /// -/// The central-side gRPC ingress for a node's deployment ack (ClusterClient→gRPC migration +/// The central-side gRPC ingress for a node's deployment ack (mesh gRPC command plane /// Phase 1B). It decodes the wire message and hands it to the registered -/// CentralCommunicationActor — the same actor the ClusterClient receptionist delivered to, +/// CentralCommunicationActor — the same actor the DistributedPubSub leg delivers to, /// whose forward-to-the-coordinator path is untouched. /// /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs index 283ad7f8..0020c7ff 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs @@ -471,7 +471,7 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase { // 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. + // nothing on the node side announces itself; the DB row is the only source. // // 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 diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs index f5fd9776..2d110eb6 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshGrpcCommandDialerTests.cs @@ -13,7 +13,7 @@ using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; /// -/// ClusterClient→gRPC Phase 1A — central's outbound command dial. One node, one call, no retry. +/// mesh gRPC command plane, Phase 1A — central's outbound command dial. One node, one call, no retry. /// /// /// The facts that matter are the credential ones: the bearer must be the TARGET node's cluster's diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderTests.cs index 3360ab9f..6ccafb65 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderTests.cs @@ -11,7 +11,7 @@ using ZB.MOM.WW.Secrets.Abstractions; namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication; /// -/// ClusterClient→gRPC Phase 0 — central's per-cluster PSK lookup. Central talks to N site +/// mesh gRPC command plane, Phase 0 — central's per-cluster PSK lookup. Central talks to N site /// clusters, each with its own key, so every outbound dial and every inbound ack verification /// goes through here. /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs index 9751788f..0e65d371 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshCommActorPathTests.cs @@ -10,11 +10,12 @@ namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; /// /// /// -/// /user/central-communication and /user/node-communication are the wire -/// contract of the central↔node boundary: central sends to a path string, and the node -/// registers under one. Nothing else in the codebase asserts they agree. Renaming either -/// actor would compile, pass every unit test, deploy — and deliver nothing, with no error -/// on either side, because a ClusterClient send to an unregistered path is simply dropped. +/// /user/central-communication and /user/node-communication are where each +/// side's gRPC ingress expects to find its comm actor: the services resolve them by registry +/// key, and these constants are the names the actors are spawned under. Nothing else in the +/// codebase asserts they agree. Renaming either actor would compile, pass every unit test, +/// deploy — and refuse every inbound call as Unavailable ("the node is still +/// starting"), which reads as a boot-window race rather than a permanent misnaming. /// /// /// Booting the real host is the point. A test that spawned the actors itself would pin the @@ -30,8 +31,8 @@ public class MeshCommActorPathTests .Ask(new Identify(path), TimeSpan.FromSeconds(10)); identity.Subject.ShouldNotBeNull( - $"{path} is the ClusterClient wire contract for the central↔node boundary; nothing else " - + "pins it, and a mismatch delivers nothing without erroring on either side"); + $"{path} is where the central↔node boundary expects its comm actor; nothing else pins " + + "it, and a mismatch turns every inbound call into a misleading readiness rejection"); } [Fact] @@ -60,7 +61,7 @@ public class MeshCommActorPathTests // 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 + // NOT MeshGrpcBoundaryTests: that class proves a command only arrives if the target is + // registered under its registry key, which is necessary but not discriminating — with one node // per side, per-node and singleton registration look identical. } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshGrpcBoundaryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshGrpcBoundaryTests.cs new file mode 100644 index 00000000..73564688 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/MeshGrpcBoundaryTests.cs @@ -0,0 +1,422 @@ +using System.Diagnostics.CodeAnalysis; +using System.Net; +using Akka.Actor; +using Akka.Cluster; +using Akka.Configuration; +using Akka.Hosting; +using Grpc.Core; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; +using ZB.MOM.WW.OtOpcUa.Host.Grpc; +using ZB.MOM.WW.OtOpcUa.Runtime; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; +using AkkaCluster = Akka.Cluster.Cluster; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// Drives the mesh command plane across two genuinely separate s over +/// a real gRPC wire — the only test in the repo where the central↔node boundary carries a +/// message off-process. +/// +/// +/// +/// Replaces MeshClusterClientBoundaryTests, which did the same job for the Akka +/// ClusterClient transport that this program deleted. The reason it exists is unchanged: +/// a fake dialer tests actor logic and cannot fail for any transport reason — not a +/// service that was never mapped, not an interceptor that refuses the call, not a +/// serialization break. Those are the failure modes this class exercises, end to end +/// through the REAL CentralCommunicationActor, MeshGrpcCommandDialer, +/// MeshCommandAuthInterceptor, NodeCommandGrpcService and +/// NodeCommunicationActor. +/// +/// +/// Three clusters, one ActorSystem name, joined ONLY by gRPC. The shared name is kept +/// from the predecessor so the shape is comparable; each system seeds itself and none lists +/// another, and +/// +/// asserts the separation outright. That assertion is the whole point of the program — the +/// cross-WAN Akka association is gone — so it is kept as a test rather than left to a live +/// gate: if the three systems ever gossiped into one mesh, every delivery below could be +/// plain local messaging and the class would be testing nothing. +/// +/// +/// Per-cluster keys, not one fleet key. Each site node's interceptor is configured +/// with its OWN MeshTransport:GrpcPsk and central's provider hands the dialer the key +/// for the cluster it is dialling. A fan that presented the wrong cluster's key would be +/// refused at the far side, so the happy path below also pins that the dialer selects by +/// ClusterId. +/// +/// +public sealed class MeshGrpcBoundaryTests : IAsyncLifetime +{ + private const string SharedSystemName = "otopcua"; + private const string SiteAKey = "psk-site-a"; + private const string SiteBKey = "psk-site-b"; + + private static readonly TimeSpan Arrival = TimeSpan.FromSeconds(30); + + /// + /// How long the falsifiability control waits before concluding nothing arrived. Deliberately + /// shorter than but far longer than a successful delivery takes in + /// practice, so a pass means "the boundary refused it", not "the clock ran out first". + /// + private static readonly TimeSpan Silence = TimeSpan.FromSeconds(10); + + private ActorSystem _central = null!; + private SiteNode _siteA = null!; + private SiteNode _siteB = null!; + + /// + public async ValueTask InitializeAsync() + { + _central = StartSelfSeededSystem(); + _siteA = await SiteNode.StartAsync("SITE-A", SiteAKey); + _siteB = await SiteNode.StartAsync("SITE-B", SiteBKey); + + await Task.WhenAll( + AwaitSingleMemberUpAsync(_central), + AwaitSingleMemberUpAsync(_siteA.System), + AwaitSingleMemberUpAsync(_siteB.System)); + } + + /// + public async ValueTask DisposeAsync() + { + await _siteA.DisposeAsync(); + await _siteB.DisposeAsync(); + await ShutdownAsync(_central); + } + + [Fact] + public void No_cross_mesh_Akka_association_forms_between_central_and_either_site() + { + // THE assertion of the ClusterClient→gRPC program, kept as a test rather than a live-gate + // note. Three systems sharing a name in one process are one gossip accident away from being + // one cluster, at which point the gRPC boundary is no longer the thing under test — and, + // more importantly, the cross-WAN Akka association this program removed would be back. + var central = AkkaCluster.Get(_central); + var siteA = AkkaCluster.Get(_siteA.System); + var siteB = AkkaCluster.Get(_siteB.System); + + central.State.Members.Count.ShouldBe(1, + "central must be its own single-member cluster; commands cross to a site over gRPC only"); + siteA.State.Members.Count.ShouldBe(1, "site-A must be its own single-member cluster"); + siteB.State.Members.Count.ShouldBe(1, "site-B must be its own single-member cluster"); + + // Stated separately from the counts: a count of 1 on each side would also hold in the instant + // before a join completed, whereas naming the addresses says no side ever learned another's. + foreach (var (name, cluster) in new[] + { + ("central", central), ("site-A", siteA), ("site-B", siteB), + }) + { + var others = new[] { central.SelfAddress, siteA.SelfAddress, siteB.SelfAddress } + .Where(a => a != cluster.SelfAddress); + foreach (var other in others) + { + cluster.State.Members.ShouldNotContain( + m => m.Address == other, + $"{name} must not have {other} as a cluster member — the cross-mesh Akka " + + "association is exactly what the gRPC command plane replaced"); + } + } + } + + [Fact] + public async Task One_dispatch_through_central_reaches_a_node_in_every_cluster_mesh() + { + // The direct counterpart of the deleted ClusterClient fan-out test, at the transport level and + // driving the REAL CentralCommunicationActor. Two genuinely separate site meshes, each with + // its own gRPC command endpoint and its own per-cluster key; one MeshCommand must reach BOTH. + var dbFactory = SeedTwoClusterDb(_siteA, _siteB); + var actor = _central.ActorOf(CentralCommunicationActor.Props( + dbFactory, + GrpcOptions(), + () => null, + () => Dialer(), + () => new FixedPskProvider())); + + var command = new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid()); + var deadline = DateTime.UtcNow + Arrival; + while (DateTime.UtcNow < deadline + && !(_siteA.Inbox.Task.IsCompleted && _siteB.Inbox.Task.IsCompleted)) + { + // Re-sent rather than sent once: the actor's first contact load is a piped DB read, and a + // command told before it lands is dropped as "no targets" by design. Re-sending weakens + // nothing — the wrong-key control below re-sends on the same schedule and must still + // receive nothing. + actor.Tell(new MeshCommand("deployments", command)); + await Task.Delay(250, TestContext.Current.CancellationToken); + } + + _siteA.Inbox.Task.IsCompleted.ShouldBeTrue( + "site-A's mesh must receive the dispatch on its own gRPC command endpoint"); + _siteB.Inbox.Task.IsCompleted.ShouldBeTrue( + "site-B's mesh must receive the SAME dispatch on its own endpoint, under its own " + + "per-cluster key — a fan that stopped at the first cluster would leave site-B silently " + + "on its old configuration while the deployment sealed green"); + + (await _siteA.Inbox.Task).ShouldBeOfType().DriverInstanceId.ShouldBe("driver-1"); + (await _siteB.Inbox.Task).ShouldBeOfType().DriverInstanceId.ShouldBe("driver-1"); + } + + [Fact] + public async Task Falsifiability_control_a_dial_carrying_the_wrong_cluster_key_delivers_nothing() + { + // Without this, the test above cannot distinguish "the gRPC boundary carried the command" + // from "something else in the process delivered it". Same wire, same service, same comm + // actor — and with a key site-A's interceptor does not accept, it must NOT arrive. + var envelope = MeshTransportDtoMapper.ToEnvelope( + new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid())); + var dialer = Dialer(new FixedPskProvider("not-the-key")); + + var deadline = DateTime.UtcNow + Silence; + while (DateTime.UtcNow < deadline && !_siteA.Inbox.Task.IsCompleted) + { + try + { + await dialer.DispatchAsync( + _siteA.Endpoint, envelope, "SITE-A", TestContext.Current.CancellationToken); + } + catch (RpcException) + { + // Expected: the node's MeshCommandAuthInterceptor refuses the bearer. The assertion + // is what did NOT arrive, so the rejection itself is swallowed rather than asserted + // on here — MeshCommandAuthInterceptorTests owns the status-code contract. + } + + await Task.Delay(500, TestContext.Current.CancellationToken); + } + + _siteA.Inbox.Task.IsCompleted.ShouldBeFalse( + "delivery survived presenting the wrong per-cluster key, so the positive test above is " + + "not measuring the authenticated gRPC boundary at all"); + + // Same node, same dialer shape, right key: proves the silence above was the auth gate rather + // than a dead endpoint or a misaddressed call. + await Dialer().DispatchAsync( + _siteA.Endpoint, envelope, "SITE-A", TestContext.Current.CancellationToken); + await _siteA.Inbox.Task.WaitAsync(Arrival, TestContext.Current.CancellationToken); + } + + // ---- harness --------------------------------------------------------------------------------- + + private static MeshGrpcCommandDialer Dialer(IMeshPskProvider? psk = null) => + new(psk ?? new FixedPskProvider(), Options.Create(GrpcOptions()), + NullLogger.Instance); + + private static MeshTransportOptions GrpcOptions() => new() + { + Mode = MeshTransportOptions.ModeGrpc, + ContactRefreshSeconds = 1, + DispatchTimeoutSeconds = 10, + }; + + /// + /// Hands out the per-cluster key each site node's interceptor expects. Stands in for the real + /// MeshPskProvider, whose own subject (secret-store lookup, caching, the known-cluster + /// gate) is MeshPskProviderTests' and would add a KEK and a SQLite file to a class + /// whose subject is the wire. + /// + private sealed class FixedPskProvider : IMeshPskProvider + { + private readonly string? _override; + + public FixedPskProvider(string? overrideKey = null) => _override = overrideKey; + + public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk) + { + psk = _override ?? KeyFor(clusterId); + return psk is not null; + } + + public bool Matches(string clusterId, string presented) => + string.Equals(presented, KeyFor(clusterId), StringComparison.Ordinal); + + public void SetKnownClusters(IReadOnlyCollection clusterIds) { } + + public void Invalidate(string clusterId) { } + + public void InvalidateAll() { } + + private static string? KeyFor(string clusterId) => clusterId switch + { + "SITE-A" => SiteAKey, + "SITE-B" => SiteBKey, + _ => null, + }; + } + + private static IDbContextFactory SeedTwoClusterDb(SiteNode a, SiteNode b) + { + var factory = new InMemoryConfigDbFactory(Guid.NewGuid().ToString("N")); + using var db = factory.CreateDbContext(); + db.ClusterNodes.Add(NodeRow("SITE-A", "node-a", a.Port)); + db.ClusterNodes.Add(NodeRow("SITE-B", "node-b", b.Port)); + db.SaveChanges(); + return factory; + } + + private static ClusterNode NodeRow(string clusterId, string nodeId, int grpcPort) => new() + { + NodeId = nodeId, + ClusterId = clusterId, + Host = "127.0.0.1", + AkkaPort = 4053, + GrpcPort = grpcPort, + ApplicationUri = $"urn:{nodeId}", + Enabled = true, + MaintenanceMode = false, + CreatedBy = "test", + }; + + private sealed class InMemoryConfigDbFactory(string dbName) + : IDbContextFactory + { + public OtOpcUaConfigDbContext CreateDbContext() + { + var opts = new DbContextOptionsBuilder() + .UseInMemoryDatabase(dbName) + .Options; + return new OtOpcUaConfigDbContext(opts); + } + } + + /// + /// One site: its own single-member cluster, its own real NodeCommunicationActor, and a + /// real h2c Kestrel listener serving the real NodeCommandGrpcService behind the real + /// MeshCommandAuthInterceptor — the production shape of a driver node's command port. + /// + private sealed class SiteNode : IAsyncDisposable + { + private WebApplication _server = null!; + + public ActorSystem System { get; private set; } = null!; + + public int Port { get; private set; } + + public string Endpoint => $"http://127.0.0.1:{Port}"; + + /// Completes with the first command re-emitted on this node's EventStream. + public TaskCompletionSource Inbox { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static async Task StartAsync(string clusterId, string psk) + { + var node = new SiteNode { System = StartSelfSeededSystem(), Port = GetFreePort() }; + + // The REAL comm actor, under the registry key the gRPC service resolves. Nothing else + // makes it reachable — there is no receptionist registration any more. + var comm = node.System.ActorOf( + NodeCommunicationActor.Props(), MeshPaths.NodeCommunicationName); + node.System.EventStream.Subscribe( + node.System.ActorOf(CaptureActor.Props(node.Inbox)), typeof(RestartDriver)); + + var registry = new ActorRegistry(); + registry.Register(comm); + + var builder = WebApplication.CreateBuilder(); + builder.Environment.ApplicationName = typeof(MeshGrpcBoundaryTests).Assembly.GetName().Name!; + builder.Logging.ClearProviders(); + builder.Services.AddSingleton(registry); + builder.Services.Configure(o => + { + o.Mode = MeshTransportOptions.ModeGrpc; + o.GrpcPsk = psk; + }); + builder.Services.AddGrpc(o => o.Interceptors.Add()); + builder.WebHost.ConfigureKestrel(k => + k.ListenAnyIP(node.Port, o => o.Protocols = HttpProtocols.Http2)); + + node._server = builder.Build(); + node._server.MapGrpcService(); + await node._server.StartAsync(TestContext.Current.CancellationToken); + + _ = clusterId; // Named for readability at the call site; the key IS the cluster boundary. + return node; + } + + public async ValueTask DisposeAsync() + { + await _server.StopAsync(TimeSpan.FromSeconds(10)); + await _server.DisposeAsync(); + await ShutdownAsync(System); + } + } + + /// Starts a single-member cluster on a dynamic remoting port. + /// + /// 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. + /// + /// The started system. + private static ActorSystem StartSelfSeededSystem() + { + 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") + .WithFallback(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"); + } + + private static async Task ShutdownAsync(ActorSystem? system) + { + if (system is null) return; + await system.Terminate().WaitAsync(TimeSpan.FromSeconds(15)); + } + + private static int GetFreePort() + { + using var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + /// Completes an inbox with the first message it receives. + private sealed class CaptureActor : ReceiveActor + { + public static Props Props(TaskCompletionSource sink) => + Akka.Actor.Props.Create(() => new CaptureActor(sink)); + + public CaptureActor(TaskCompletionSource sink) => ReceiveAny(m => sink.TrySetResult(m)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs index 510a5dbc..e04252c9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/CentralCommandAuthInterceptorTests.cs @@ -9,7 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Host.Configuration; namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration; /// -/// ClusterClient→gRPC Phase 0 — the central-side ack endpoint's inbound gate. Mapping +/// mesh gRPC command plane, Phase 0 — the central-side ack endpoint's inbound gate. Mapping /// CentralCommandService onto the pipeline is a Phase 1 task; this test covers only the /// interceptor's own authorization behavior. /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs index b96e28b0..596d36df 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/MeshCommandAuthInterceptorTests.cs @@ -9,7 +9,7 @@ using ZB.MOM.WW.OtOpcUa.Host.Configuration; namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration; /// -/// ClusterClient→gRPC Phase 0 — the node-side command endpoint's inbound gate. Mapping +/// mesh gRPC command plane, Phase 0 — the node-side command endpoint's inbound gate. Mapping /// NodeCommandService onto the pipeline is a Phase 1 task; this test covers only the /// interceptor's own authorization behavior. /// 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 index 0ccd6ea8..a6a15050 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs @@ -16,10 +16,10 @@ 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 +/// Verifies the node-side command ingress (mesh gRPC command plane, 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 +/// registered NodeCommunicationActor as the same CLR record every other transport +/// delivers, an undecodable envelope is refused with , and /// a call landing in the boot window before the actor is registered is refused with /// . /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs index bd6603f8..a8820cf1 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs @@ -13,21 +13,21 @@ using ZB.MOM.WW.OtOpcUa.Runtime.Communication; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication; /// -/// The node-side gRPC ack sender (ClusterClient→gRPC migration Task 1B.2). The client is a fake, +/// The node-side gRPC ack sender (mesh gRPC command plane, Task 1B.2). The client is a fake, /// so what is under test is the sender's own logic: the ordered single-pass over /// CentralGrpcEndpoints, the per-call credentials, and the drop that used to be silent. /// /// /// /// The load-bearing fact is that ONE endpoint is dialled on success. This replaces an -/// Akka ClusterClient.Send, which is single-target at-most-once — not a +/// one ack per node per dispatch, single-target at-most-once — not a /// SendToAll. Both central nodes route an ack to the same /// ConfigPublishCoordinator singleton proxy, whose NodeId-keyed upsert absorbs a /// duplicate idempotently — so the single-endpoint rule is defence in depth and a bound /// on wasted dials, not what protects the seal. /// /// -/// The failure side matters just as much: the receptionist dropped an undeliverable ack in +/// The failure side matters just as much: an undeliverable ack is dropped in /// silence, and the whole point of the migration is that the same drop is now a Warning naming /// every endpoint and the status it answered with. ///