diff --git a/docs/plans/2026-07-22-clusterclient-to-grpc-plan.md b/docs/plans/2026-07-22-clusterclient-to-grpc-plan.md
index dd399b4b..77a52dcc 100644
--- a/docs/plans/2026-07-22-clusterclient-to-grpc-plan.md
+++ b/docs/plans/2026-07-22-clusterclient-to-grpc-plan.md
@@ -317,7 +317,7 @@ Critical path ≈ 1B: **~4–6 weeks total**, matching the design estimate.
**Phase 1B — site command plane** (worktree, `feat/grpc-sitecommand`)
- [x] T1B.1 `site_command.proto` (6 oneof RPCs / 28 commands) + `SiteCommandDtoMapper` + round-trip golden tests (all 28 commands + 22 reply shapes + 18 nested types; reflection-driven coverage guard over the mapper surface and the generated oneof descriptors)
-- [x] T1B.2 `SiteCommandDispatcher` refactor (actor + new `SiteCommandGrpcService` share it; parked stays node-local)
+- [x] T1B.2 `SiteCommandDispatcher` refactor (actor + new `SiteCommandGrpcService` share it; parked stays node-local; failover ack-before-Leave via dry-run resolve + deferred `CommitLeave`; interceptor `DefaultGatedPrefixes` extended to `SiteCommandService`)
- [x] T1B.3 `ISiteCommandTransport` in `CentralCommunicationActor` (Akka extract + Grpc impl), `SitePairChannelProvider` (Site entity Grpc columns + DB refresh loop), `SiteTransport` flag default `Akka`
- [x] T1B.4 Tests: dispatcher routing ×28, actor envelope/reply plumbing, TestServer service tests, existing Communication suites green
- [ ] 1B DoD: rig central on `Grpc` for site-a proves full command matrix incl. standby parked retry; rebased on 1A; PR merged
diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommandDispatcher.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommandDispatcher.cs
new file mode 100644
index 00000000..e37456ef
--- /dev/null
+++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommandDispatcher.cs
@@ -0,0 +1,303 @@
+using Akka.Actor;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
+using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Actors;
+
+///
+/// The single routing truth for the 28 migrated central→site commands. Given one
+/// of those command records it decides which local target answers it —
+/// the Deployment Manager singleton proxy, the artifact/event-log/parked-message
+/// handlers, or the node-local failover path — preserving EXACTLY the targets and
+/// null-guard semantics used when this routing
+/// lived inline.
+///
+///
+///
+/// Two transports call this one unit: the Akka
+/// (ClusterClient) and the new SiteCommandGrpcService (gRPC). Centralising the
+/// table means the two can never drift on where a command goes. Each transport keeps
+/// its own send mechanics — the actor Forwards (preserving the Ask sender), the
+/// gRPC service Asks and encodes the reply — but both read the same
+/// here.
+///
+///
+/// The parked-message handler stays node-local on purpose. A parked retry or
+/// discard must run on the node that holds the replicated store row, so parked
+/// commands route to the per-node _parkedMessageHandler — never onto the
+/// singleton proxy (design §7.3). This is the same target the actor used; the
+/// extraction does not "fix" it.
+///
+///
+/// Excluded by design:IntegrationCallRequest — the 29th command, dead at
+/// both ends. It never enters this dispatcher; the actor keeps its own vestigial handler
+/// for it (28 of 29 migrate).
+///
+///
+public sealed class SiteCommandDispatcher
+{
+ /// How a resolved command is delivered to its target.
+ public enum RouteDisposition
+ {
+ /// Forward/Ask and route its reply back.
+ Forward,
+
+ ///
+ /// Tell and answer immediately with
+ /// — the fire-and-forget path (only
+ /// , which the downstream never acks).
+ /// The actor Forwards it and sends nothing back; the gRPC transport Tells and
+ /// returns the synthetic ack so a unary RPC still answers.
+ ///
+ TellFireAndForget,
+
+ ///
+ /// No target is available (a null-guarded handler is unregistered); answer with the
+ /// synthetic — the "handler not available" reply the actor
+ /// produced inline.
+ ///
+ ImmediateReply
+ }
+
+ /// A resolved routing decision for one command.
+ /// How the command is delivered.
+ /// The target actor for /; null for an immediate reply.
+ /// The synthetic reply for /; null for a forward.
+ public readonly record struct Route(RouteDisposition Disposition, IActorRef? Target, object? Reply);
+
+ /// The outcome of preparing a failover: the ack to send now, plus the leave to run after.
+ /// The ack the caller must send back before the node leaves.
+ ///
+ /// When non-null (accepted only), invoking it performs the real graceful
+ /// Cluster.Leave. It is deliberately deferred so the transport can flush the
+ /// first — a caller reaching the very node that is about to leave still
+ /// receives the ack rather than a broken stream.
+ ///
+ public readonly record struct FailoverOutcome(SiteFailoverAck Ack, Action? CommitLeave);
+
+ private readonly string _siteId;
+ private readonly IActorRef _deploymentManagerProxy;
+
+ // (role, dryRun) -> leaving node address, or null when there is no standby.
+ // dryRun:true resolves the target WITHOUT leaving; dryRun:false performs the Leave.
+ private readonly Func _resolveFailover;
+
+ // Registered at runtime via the actor's RegisterLocalHandler flow. Written on the actor
+ // thread, read by both the actor and the gRPC service (a Kestrel thread), so volatile.
+ private volatile IActorRef? _eventLogHandler;
+ private volatile IActorRef? _parkedMessageHandler;
+ private volatile IActorRef? _artifactHandler;
+
+ /// Creates the dispatcher.
+ /// This site's identifier, stamped into synthetic replies and matched by the failover guard.
+ /// The local Deployment Manager singleton proxy — the target for all lifecycle/OPC UA/query/route commands.
+ ///
+ /// Resolves (and, when dryRun is false, performs) the graceful leave of the oldest
+ /// Up member in a role scope. Injected so tests need no real cluster.
+ ///
+ public SiteCommandDispatcher(
+ string siteId,
+ IActorRef deploymentManagerProxy,
+ Func resolveFailover)
+ {
+ ArgumentNullException.ThrowIfNull(siteId);
+ ArgumentNullException.ThrowIfNull(deploymentManagerProxy);
+ ArgumentNullException.ThrowIfNull(resolveFailover);
+
+ _siteId = siteId;
+ _deploymentManagerProxy = deploymentManagerProxy;
+ _resolveFailover = resolveFailover;
+ }
+
+ /// Registers the site event-log query handler (a cluster singleton proxy).
+ /// The event-log handler.
+ public void RegisterEventLogHandler(IActorRef handler) => _eventLogHandler = handler;
+
+ /// Registers the node-local parked-message handler (the replicated-store owner on this node).
+ /// The parked-message handler.
+ public void RegisterParkedMessageHandler(IActorRef handler) => _parkedMessageHandler = handler;
+
+ /// Registers the artifact-deployment handler.
+ /// The artifact handler.
+ public void RegisterArtifactHandler(IActorRef handler) => _artifactHandler = handler;
+
+ ///
+ /// Resolves how one of the 27 non-failover commands is delivered.
+ /// is handled separately via / because its
+ /// leave is deferred; passing it here throws.
+ ///
+ /// The command to route.
+ /// The routing decision.
+ /// The command is not a migrated site command (or is failover).
+ public Route ResolveRoute(object command)
+ {
+ ArgumentNullException.ThrowIfNull(command);
+
+ return command switch
+ {
+ // ── Deployment + instance lifecycle → Deployment Manager singleton proxy ──
+ RefreshDeploymentCommand => ToProxy(),
+ EnableInstanceCommand => ToProxy(),
+ DisableInstanceCommand => ToProxy(),
+ DeleteInstanceCommand => ToProxy(),
+ DeploymentStateQueryRequest => ToProxy(),
+
+ // ── Artifact deployment → artifact handler (null-guarded) ──
+ DeployArtifactsCommand c => _artifactHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ArtifactDeploymentResponse(
+ c.DeploymentId, _siteId, false, "Artifact handler not available", DateTimeOffset.UtcNow)),
+
+ // ── Interactive OPC UA / MxGateway → Deployment Manager singleton proxy ──
+ // The singleton always lands on the active node, which owns the live sessions.
+ BrowseNodeCommand => ToProxy(),
+ SearchAddressSpaceCommand => ToProxy(),
+ ReadTagValuesCommand => ToProxy(),
+ VerifyEndpointCommand => ToProxy(),
+ TrustServerCertCommand => ToProxy(),
+ ListServerCertsCommand => ToProxy(),
+ RemoveServerCertCommand => ToProxy(),
+ WriteTagRequest => ToProxy(),
+
+ // ── Remote queries: event log (null-guarded) + debug view (proxy) ──
+ EventLogQueryRequest r => _eventLogHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new EventLogQueryResponse(
+ r.CorrelationId, _siteId, [], null, false, false,
+ "Event log handler not available", DateTimeOffset.UtcNow)),
+ DebugSnapshotRequest => ToProxy(),
+ SubscribeDebugViewRequest => ToProxy(),
+ // Fire-and-forget: the Deployment Manager never acks an unsubscribe, so the gRPC
+ // transport Tells it and returns the synthetic ack; the actor just Forwards.
+ UnsubscribeDebugViewRequest => new Route(
+ RouteDisposition.TellFireAndForget, _deploymentManagerProxy, UnsubscribeDebugViewAck.Instance),
+
+ // ── Parked store-and-forward actions → node-local parked handler (null-guarded) ──
+ ParkedMessageQueryRequest r => _parkedMessageHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ParkedMessageQueryResponse(
+ r.CorrelationId, _siteId, [], 0, r.PageNumber, r.PageSize, false,
+ "Parked message handler not available", DateTimeOffset.UtcNow)),
+ ParkedMessageRetryRequest r => _parkedMessageHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ParkedMessageRetryResponse(
+ r.CorrelationId, false, "Parked message handler not available")),
+ ParkedMessageDiscardRequest r => _parkedMessageHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ParkedMessageDiscardResponse(
+ r.CorrelationId, false, "Parked message handler not available")),
+ RetryParkedOperation r => _parkedMessageHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ParkedOperationActionAck(
+ r.CorrelationId, Applied: false, "Parked message handler not available")),
+ DiscardParkedOperation r => _parkedMessageHandler is { } h
+ ? Forwarded(h)
+ : Immediate(new ParkedOperationActionAck(
+ r.CorrelationId, Applied: false, "Parked message handler not available")),
+
+ // ── Inbound-API Route.To() relays → Deployment Manager singleton proxy ──
+ RouteToCallRequest => ToProxy(),
+ RouteToGetAttributesRequest => ToProxy(),
+ RouteToSetAttributesRequest => ToProxy(),
+ RouteToWaitForAttributeRequest => ToProxy(),
+
+ TriggerSiteFailover => throw new ArgumentException(
+ "TriggerSiteFailover is handled by PrepareFailover/HandleFailover, not ResolveRoute.",
+ nameof(command)),
+
+ _ => throw new ArgumentException(
+ $"'{command.GetType().Name}' is not a migrated site command.", nameof(command))
+ };
+
+ Route ToProxy() => Forwarded(_deploymentManagerProxy);
+ }
+
+ private static Route Forwarded(IActorRef target) => new(RouteDisposition.Forward, target, null);
+
+ private static Route Immediate(object reply) => new(RouteDisposition.ImmediateReply, null, reply);
+
+ ///
+ /// The actor's failover path: resolve the standby AND issue the leave in one step (today's
+ /// coupled behaviour over ClusterClient, where Tell merely enqueues the ack so leave
+ /// order is immaterial), then return the ack.
+ ///
+ /// The failover command.
+ /// The ack to send back.
+ public SiteFailoverAck HandleFailover(TriggerSiteFailover msg)
+ => FailoverCore(msg, commitLeaveImmediately: true).Ack;
+
+ ///
+ /// The gRPC failover path: resolve the standby WITHOUT leaving, build the ack, and hand back a
+ /// deferred . The caller must send the ack before
+ /// invoking the leave — otherwise a caller reaching the leaving node sees a broken stream
+ /// instead of its ack (the ack-before-Leave rule).
+ ///
+ /// The failover command.
+ /// The ack and the deferred leave (leave is null when refused).
+ public FailoverOutcome PrepareFailover(TriggerSiteFailover msg)
+ => FailoverCore(msg, commitLeaveImmediately: false);
+
+ private FailoverOutcome FailoverCore(TriggerSiteFailover msg, bool commitLeaveImmediately)
+ {
+ ArgumentNullException.ThrowIfNull(msg);
+
+ // A misrouted command must be refused, not silently acted on — acting would fail over a
+ // site the operator never selected. Checked before resolving so the resolver is untouched.
+ if (!string.Equals(msg.SiteId, _siteId, StringComparison.Ordinal))
+ {
+ return new FailoverOutcome(
+ new SiteFailoverAck(
+ msg.CorrelationId, Accepted: false, TargetAddress: null,
+ ErrorMessage: $"Command addressed to site '{msg.SiteId}' but this node serves '{_siteId}'."),
+ CommitLeave: null);
+ }
+
+ // Site singletons are scoped to the site-specific role, so failover must target that role.
+ var role = $"site-{_siteId}";
+ try
+ {
+ if (commitLeaveImmediately)
+ {
+ var target = _resolveFailover(role, false);
+ if (target is null)
+ {
+ return new FailoverOutcome(NoPeerAck(msg), CommitLeave: null);
+ }
+
+ return new FailoverOutcome(
+ new SiteFailoverAck(msg.CorrelationId, Accepted: true, target, ErrorMessage: null),
+ CommitLeave: null);
+ }
+
+ var resolved = _resolveFailover(role, true);
+ if (resolved is null)
+ {
+ return new FailoverOutcome(NoPeerAck(msg), CommitLeave: null);
+ }
+
+ return new FailoverOutcome(
+ new SiteFailoverAck(msg.CorrelationId, Accepted: true, resolved, ErrorMessage: null),
+ CommitLeave: () => _resolveFailover(role, false));
+ }
+ catch (Exception ex)
+ {
+ // A fault must be reported to the operator, never thrown into supervision (over
+ // ClusterClient) or surfaced as a broken stream (over gRPC).
+ return new FailoverOutcome(
+ new SiteFailoverAck(
+ msg.CorrelationId, Accepted: false, TargetAddress: null, ErrorMessage: ex.Message),
+ CommitLeave: null);
+ }
+ }
+
+ private static SiteFailoverAck NoPeerAck(TriggerSiteFailover msg) => new(
+ msg.CorrelationId, Accepted: false, TargetAddress: null,
+ ErrorMessage: "No standby available — failing over a lone node would be an outage.");
+}
diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs
index b8a2a6f0..203f5094 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs
@@ -36,13 +36,20 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
/// do not need a real cluster.
///
private readonly Func _isActiveCheck;
- private readonly Func _failOverRole;
///
/// Reference to the local Deployment Manager singleton proxy.
///
private readonly IActorRef _deploymentManagerProxy;
+ ///
+ /// The single routing truth for the 28 migrated central→site commands, shared with
+ /// the gRPC SiteCommandGrpcService so the two transports cannot drift. In
+ /// production it is created by the Host and passed in (so the gRPC server holds the
+ /// same instance); when unset (tests) the actor builds its own from the failover seam.
+ ///
+ private readonly SiteCommandDispatcher _dispatcher;
+
///
/// The site→central transport. Finalized in to the injected instance,
/// or a default (ClusterClient) when none is supplied — so
@@ -54,13 +61,10 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
private readonly ICentralTransport? _injectedTransport;
///
- /// Local actor references for routing specific message patterns.
- /// Populated via registration messages.
+ /// Handler for the vestigial — the one command NOT
+ /// migrated to the dispatcher (dead at both ends), so it still routes on the actor.
///
- private IActorRef? _eventLogHandler;
- private IActorRef? _parkedMessageHandler;
private IActorRef? _integrationHandler;
- private IActorRef? _artifactHandler;
/// Akka timer scheduler injected by the framework via .
public ITimerScheduler Timers { get; set; } = null!;
@@ -82,24 +86,41 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
/// Host injects a when
/// ScadaBridge:Communication:CentralTransport is Grpc.
///
+ ///
+ /// The shared (production: created by the Host and also
+ /// handed to the gRPC command service, so both transports route through one instance).
+ /// null makes the actor build its own from — the shape
+ /// the existing tests use.
+ ///
public SiteCommunicationActor(
string siteId,
CommunicationOptions options,
IActorRef deploymentManagerProxy,
Func? isActiveCheck = null,
Func? failOverRole = null,
- ICentralTransport? transport = null)
+ ICentralTransport? transport = null,
+ SiteCommandDispatcher? dispatcher = null)
{
_siteId = siteId;
_options = options;
_deploymentManagerProxy = deploymentManagerProxy;
_isActiveCheck = isActiveCheck ?? DefaultIsActiveCheck;
- _failOverRole = failOverRole ?? DefaultFailOverRole;
_injectedTransport = transport;
// Finalized in PreStart (where _log is usable for the default transport); assigned here
// too so the field is definitely-assigned for the constructor's Receive closures.
_transport = transport!;
+ // When no shared dispatcher is supplied, build one over the same failover seam the
+ // actor used before extraction: an injected Func (tests) that both resolves and leaves,
+ // or the shared ClusterFailoverCoordinator. The actor only ever commits the leave
+ // immediately (dryRun:false), so an injected coupled Func fits unchanged.
+ var system = Context.System;
+ Func resolveFailover = failOverRole is not null
+ ? (role, _) => failOverRole(role)
+ : (role, dryRun) =>
+ ClusterState.ClusterFailoverCoordinator.FailOverOldest(system, role, dryRun)?.ToString();
+ _dispatcher = dispatcher ?? new SiteCommandDispatcher(siteId, deploymentManagerProxy, resolveFailover);
+
// Registration. Feeding the ClusterClient into the transport is a no-op unless the
// default/Akka transport is in use — the gRPC transport dials configured endpoints and
// never receives this message (the Host does not create a ClusterClient for it).
@@ -109,37 +130,44 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
});
Receive(HandleRegisterLocalHandler);
- // Pattern 1: Instance Deployment — forward to Deployment Manager
- Receive(msg =>
- {
- _log.Debug("Routing RefreshDeploymentCommand for {0} to DeploymentManager", msg.InstanceUniqueName);
- _deploymentManagerProxy.Forward(msg);
- });
+ // ── The 27 migrated central→site commands (28th is failover, below) all route
+ // through the shared SiteCommandDispatcher — the single routing truth also used by
+ // the gRPC SiteCommandGrpcService. The actor's job per command is unchanged: Forward
+ // to the resolved target (preserving the central Ask sender so replies route straight
+ // back to the waiting Ask), or Tell the caller the dispatcher's synthetic reply when a
+ // null-guarded handler is unregistered. See SiteCommandDispatcher for the target of
+ // each command and why the parked handler stays node-local.
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
+ Receive(cmd => DispatchCommand(cmd));
- // Pattern 2: Lifecycle — forward to Deployment Manager
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Query-the-site-before-redeploy — forward to
- // the Deployment Manager, which owns the deployed-config store and
- // answers with the instance's currently-applied deployment identity.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Pattern 3: Artifact Deployment — forward to artifact handler if registered
- Receive(msg =>
- {
- if (_artifactHandler != null)
- _artifactHandler.Forward(msg);
- else
- {
- _log.Warning("No artifact handler registered, replying with failure");
- Sender.Tell(new ArtifactDeploymentResponse(
- msg.DeploymentId, _siteId, false, "Artifact handler not available", DateTimeOffset.UtcNow));
- }
- });
-
- // Pattern 4: Integration Routing — forward to integration handler
+ // Integration Routing — the 29th command, NOT migrated to the dispatcher (dead at
+ // both ends; no production code registers the handler). Kept on the actor so the
+ // dispatcher's command surface stays the 28 that actually migrate.
Receive(msg =>
{
if (_integrationHandler != null)
@@ -151,143 +179,15 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
}
});
- // Pattern 5: Debug View — forward to Deployment Manager (which routes to Instance Actor)
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Pattern 6a: Debug Snapshot (one-shot) — forward to Deployment Manager
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Inbound API Route.To() — forward to Deployment Manager for instance routing
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // OPC UA Tag Browser (interactive design-time query) — forward to the
- // Deployment Manager singleton, which always lands on the active site
- // node. Routing to the site-local /user/dcl-manager directly is wrong
- // because the standby node has a dcl-manager too, but its
- // DataConnectionActor children (which own the live OPC UA sessions)
- // only exist on the singleton's node. The singleton then re-forwards
- // to its own /user/dcl-manager, which DOES have the connection.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Test Bindings (interactive design-time read) — same routing rationale
- // as BrowseNodeCommand above: the singleton always lands on the
- // active site node, which is the node that owns the DataConnectionActor
- // children holding the live OPC UA sessions.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // OPC UA tag-picker address-space search and secured-write execute
- // — same singleton routing rationale as BrowseNodeCommand above: the
- // DataConnectionActor children that own the live OPC UA sessions exist only
- // on the singleton's (active) node, so these must hop through the Deployment
- // Manager proxy too. Without these forwards the commands dead-letter and the
- // central Ask times out. Forward preserves the central Ask sender so the
- // result routes straight back to the waiting Ask.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // OPC UA endpoint Verify — probes a (possibly unsaved) endpoint config
- // WITHOUT persisting it. The Deployment Manager singleton's dcl-manager runs
- // the probe directly (no existing connection required), so — like the
- // commands above — Verify routes through the singleton's active node.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // OPC UA server-certificate trust management — forward to the
- // Deployment Manager singleton, which owns the cross-node trust broadcast.
- // The trusted-peer PKI store is node-wide per site node, so a trust/remove
- // decision must reach BOTH nodes' CertStoreActor; the singleton broadcasts
- // to every site node (list answers from the singleton's own node). The
- // singleton always lands on the active node, the same routing rationale as
- // BrowseNodeCommand above. Forward preserves the central Ask sender so the
- // CertTrustResult routes straight back to the waiting Ask.
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
- Receive(msg => _deploymentManagerProxy.Forward(msg));
-
- // Pattern 7: Remote Queries
- Receive(msg =>
- {
- if (_eventLogHandler != null)
- _eventLogHandler.Forward(msg);
- else
- {
- Sender.Tell(new EventLogQueryResponse(
- msg.CorrelationId, _siteId, [], null, false, false,
- "Event log handler not available", DateTimeOffset.UtcNow));
- }
- });
-
- Receive(msg =>
- {
- if (_parkedMessageHandler != null)
- _parkedMessageHandler.Forward(msg);
- else
- {
- Sender.Tell(new ParkedMessageQueryResponse(
- msg.CorrelationId, _siteId, [], 0, msg.PageNumber, msg.PageSize, false,
- "Parked message handler not available", DateTimeOffset.UtcNow));
- }
- });
-
- Receive(msg =>
- {
- if (_parkedMessageHandler != null)
- _parkedMessageHandler.Forward(msg);
- else
- {
- Sender.Tell(new ParkedMessageRetryResponse(
- msg.CorrelationId, false, "Parked message handler not available"));
- }
- });
-
- Receive(msg =>
- {
- if (_parkedMessageHandler != null)
- _parkedMessageHandler.Forward(msg);
- else
- {
- Sender.Tell(new ParkedMessageDiscardResponse(
- msg.CorrelationId, false, "Parked message handler not available"));
- }
- });
-
- // Central→site Retry/Discard relay for parked cached
- // operations. SiteCallAuditActor relays these over the command/control
- // channel; the parked-message handler executes them against the local
- // S&F buffer and replies a ParkedOperationActionAck that routes back to
- // the relaying SiteCallAuditActor's Ask.
- Receive(msg =>
- {
- if (_parkedMessageHandler != null)
- _parkedMessageHandler.Forward(msg);
- else
- {
- Sender.Tell(new ParkedOperationActionAck(
- msg.CorrelationId, Applied: false, "Parked message handler not available"));
- }
- });
-
- Receive(msg =>
- {
- if (_parkedMessageHandler != null)
- _parkedMessageHandler.Forward(msg);
- else
- {
- Sender.Tell(new ParkedOperationActionAck(
- msg.CorrelationId, Applied: false, "Parked message handler not available"));
- }
- });
-
// Central→site manual failover relay. Central and the site are separate clusters,
// so central can only ask — this node performs the graceful Leave locally, scoped to
// the SITE-SPECIFIC role, because that is what site singletons (the Deployment
// Manager) are placed on. Either node may receive this (the actor is per-node, not a
// singleton, and contact rotation picks whichever answers); the target is resolved
- // from cluster state, not from who received the message.
- Receive(HandleTriggerSiteFailover);
+ // from cluster state, not from who received the message. Over ClusterClient the ack
+ // Tell merely enqueues, so the dispatcher resolves-and-leaves in one step (the gRPC
+ // transport defers the leave to keep ack-before-Leave — see PrepareFailover).
+ Receive(msg => Sender.Tell(_dispatcher.HandleFailover(msg)));
// The seven site→central sends now delegate to the injected transport (ClusterClient by
// default, gRPC when configured). Each handler captures the current Sender as the reply
@@ -352,21 +252,47 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
_options.ApplicationHeartbeatInterval);
}
+ ///
+ /// Executes a resolved command route within the actor: Forward to the target (preserving the
+ /// central Ask sender so the reply routes straight back to the waiting Ask), or — when a
+ /// null-guarded handler is unregistered — Tell the caller the dispatcher's synthetic reply.
+ /// The fire-and-forget disposition (UnsubscribeDebugView) is a plain Forward here, exactly as
+ /// before: over ClusterClient the site never acked it, so the synthetic ack is a gRPC-only
+ /// concern.
+ ///
+ /// The migrated central→site command to route.
+ private void DispatchCommand(object command)
+ {
+ var route = _dispatcher.ResolveRoute(command);
+ switch (route.Disposition)
+ {
+ case SiteCommandDispatcher.RouteDisposition.ImmediateReply:
+ Sender.Tell(route.Reply!);
+ break;
+ default:
+ // Forward and TellFireAndForget both Forward on the actor path.
+ route.Target!.Forward(command);
+ break;
+ }
+ }
+
private void HandleRegisterLocalHandler(RegisterLocalHandler msg)
{
+ // The migrated handlers live on the shared dispatcher so the gRPC command service sees the
+ // same registrations. Integration is the one command kept on the actor (see the receive).
switch (msg.HandlerType)
{
case LocalHandlerType.EventLog:
- _eventLogHandler = msg.Handler;
+ _dispatcher.RegisterEventLogHandler(msg.Handler);
break;
case LocalHandlerType.ParkedMessages:
- _parkedMessageHandler = msg.Handler;
+ _dispatcher.RegisterParkedMessageHandler(msg.Handler);
break;
case LocalHandlerType.Integration:
_integrationHandler = msg.Handler;
break;
case LocalHandlerType.Artifacts:
- _artifactHandler = msg.Handler;
+ _dispatcher.RegisterArtifactHandler(msg.Handler);
break;
}
@@ -431,69 +357,6 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
private bool DefaultIsActiveCheck() =>
ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(Cluster.Get(Context.System));
- ///
- /// Handles a central-initiated site failover. Refuses a command addressed to a different
- /// site (a misroute must never silently fail over a site the operator did not select) and
- /// refuses when the site pair has no peer to take over. The ack is sent BEFORE the Leave
- /// takes effect on the wire, so it still reaches central even when this node is the one
- /// leaving.
- ///
- private void HandleTriggerSiteFailover(TriggerSiteFailover msg)
- {
- if (!string.Equals(msg.SiteId, _siteId, StringComparison.Ordinal))
- {
- _log.Warning(
- "Refusing TriggerSiteFailover addressed to site {Requested}; this node serves {Actual}",
- msg.SiteId, _siteId);
- Sender.Tell(new SiteFailoverAck(
- msg.CorrelationId, Accepted: false, TargetAddress: null,
- ErrorMessage: $"Command addressed to site '{msg.SiteId}' but this node serves '{_siteId}'."));
- return;
- }
-
- var role = $"site-{_siteId}";
- try
- {
- var target = _failOverRole(role);
- if (target is null)
- {
- _log.Warning(
- "Refusing TriggerSiteFailover for {SiteId}: fewer than 2 Up members in role {Role}, "
- + "so there is no standby to take over", _siteId, role);
- Sender.Tell(new SiteFailoverAck(
- msg.CorrelationId, Accepted: false, TargetAddress: null,
- ErrorMessage: "No standby available — failing over a lone node would be an outage."));
- return;
- }
-
- _log.Warning(
- "Manual failover requested by central for site {SiteId}: {Target} is leaving the "
- + "site cluster gracefully; its singletons hand over to the standby.", _siteId, target);
- Sender.Tell(new SiteFailoverAck(msg.CorrelationId, Accepted: true, target, ErrorMessage: null));
- }
- catch (Exception ex)
- {
- // A fault here must be reported to the operator, not thrown into supervision —
- // restarting the communication actor would drop central's Ask into a timeout and
- // lose the reason.
- _log.Error(ex, "TriggerSiteFailover for {SiteId} faulted", _siteId);
- Sender.Tell(new SiteFailoverAck(
- msg.CorrelationId, Accepted: false, TargetAddress: null, ErrorMessage: ex.Message));
- }
- }
-
- ///
- /// Production failover action: gracefully Leave the oldest Up member carrying
- /// , via the shared
- /// so the central and site paths cannot drift. Injected in tests for the same reason
- /// is — a real Leave needs Akka.Cluster in the
- /// ActorSystem, which the TestKit system does not load.
- ///
- /// Site-specific role scope.
- /// Address of the leaving node, or null when there is no peer.
- private string? DefaultFailOverRole(string role) =>
- ClusterState.ClusterFailoverCoordinator.FailOverOldest(Context.System, role)?.ToString();
-
// ── Internal messages ──
internal record SendHeartbeat;
diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteCommandGrpcService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteCommandGrpcService.cs
new file mode 100644
index 00000000..b0fc482d
--- /dev/null
+++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteCommandGrpcService.cs
@@ -0,0 +1,227 @@
+using Akka.Actor;
+using Grpc.Core;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
+using ZB.MOM.WW.ScadaBridge.Communication.Actors;
+using GrpcStatus = Grpc.Core.Status;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+
+///
+/// gRPC front door for the central→site command plane: decodes a proto command (via
+/// ), routes it through the ONE
+/// the Akka SiteCommunicationActor also uses, and
+/// encodes the reply. One routing truth, two transports.
+///
+///
+///
+/// Mapped in the site branch next to SiteStreamGrpcServer, behind the same
+/// ControlPlaneAuthInterceptor PSK gate and the same readiness convention: calls are
+/// rejected with until is called
+/// once the site actor system is up (mirrors SiteStreamGrpcServer.SetReady).
+///
+///
+/// Coexistence — server-side only. This makes the site ALSO listen on gRPC for commands;
+/// nothing central flips to gRPC here (that is T1B.3). Central still dials sites via ClusterClient.
+///
+///
+/// Fire-and-forget. The Akka path never acks UnsubscribeDebugView, but a unary RPC
+/// must answer, so the dispatcher marks it
+/// and this service Tells the target then returns the synthetic ack.
+///
+///
+/// Ack-before-Leave.TriggerFailover resolves the standby WITHOUT leaving
+/// (), returns the ack, and only THEN schedules
+/// the real Cluster.Leave — so a caller reaching the very node that is about to leave still
+/// receives its ack instead of a broken stream.
+///
+///
+public sealed class SiteCommandGrpcService : SiteCommandService.SiteCommandServiceBase
+{
+ // A local Ask has no client deadline of its own; fall back to a generous ceiling when the
+ // caller set none (a deadline-less client is a test or an internal caller). When the caller
+ // DID set a gRPC deadline, honour the remaining time instead.
+ private static readonly TimeSpan DefaultAskTimeout = TimeSpan.FromMinutes(2);
+
+ private readonly ILogger _logger;
+ private readonly Action _leaveScheduler;
+
+ // Set once by SetReady after the site actor system and dispatcher exist. Read on Kestrel
+ // threads, written on the host bring-up thread — volatile.
+ private volatile SiteCommandDispatcher? _dispatcher;
+ private volatile bool _ready;
+
+ /// DI constructor.
+ /// Logger for denial/dispatch diagnostics.
+ public SiteCommandGrpcService(ILogger logger)
+ : this(logger, DeferLeaveUntilAfterReply)
+ {
+ }
+
+ ///
+ /// Test constructor letting a test observe when the deferred leave runs. Internal so DI sees a
+ /// single public constructor.
+ ///
+ /// Logger.
+ /// Runs the deferred Cluster.Leave after the ack is returned.
+ internal SiteCommandGrpcService(ILogger logger, Action leaveScheduler)
+ {
+ ArgumentNullException.ThrowIfNull(logger);
+ ArgumentNullException.ThrowIfNull(leaveScheduler);
+ _logger = logger;
+ _leaveScheduler = leaveScheduler;
+ }
+
+ ///
+ /// Marks the service ready and injects the shared routing table. Called once the site actor
+ /// system, the Deployment Manager singleton, and the local handlers are up — the same point
+ /// SiteStreamGrpcServer.SetReady is called.
+ ///
+ /// The shared dispatcher (the same instance the actor routes through).
+ public void SetReady(SiteCommandDispatcher dispatcher)
+ {
+ ArgumentNullException.ThrowIfNull(dispatcher);
+ _dispatcher = dispatcher;
+ _ready = true;
+ }
+
+ /// Whether the service is accepting commands. Exposed for tests.
+ internal bool IsReady => _ready;
+
+ ///
+ public override async Task ExecuteLifecycle(LifecycleRequest request, ServerCallContext context)
+ {
+ var command = SiteCommandDtoMapper.FromLifecycleRequest(request);
+ var reply = await DispatchAsync(command, context).ConfigureAwait(false);
+ return SiteCommandDtoMapper.ToLifecycleReply(reply);
+ }
+
+ ///
+ public override async Task ExecuteOpcUa(OpcUaRequest request, ServerCallContext context)
+ {
+ var command = SiteCommandDtoMapper.FromOpcUaRequest(request);
+ var reply = await DispatchAsync(command, context).ConfigureAwait(false);
+ return SiteCommandDtoMapper.ToOpcUaReply(reply);
+ }
+
+ ///
+ public override async Task ExecuteQuery(QueryRequest request, ServerCallContext context)
+ {
+ var command = SiteCommandDtoMapper.FromQueryRequest(request);
+ var reply = await DispatchAsync(command, context).ConfigureAwait(false);
+ return SiteCommandDtoMapper.ToQueryReply(reply);
+ }
+
+ ///
+ public override async Task ExecuteParked(ParkedRequest request, ServerCallContext context)
+ {
+ var command = SiteCommandDtoMapper.FromParkedRequest(request);
+ var reply = await DispatchAsync(command, context).ConfigureAwait(false);
+ return SiteCommandDtoMapper.ToParkedReply(reply);
+ }
+
+ ///
+ public override async Task ExecuteRoute(RouteRequest request, ServerCallContext context)
+ {
+ var command = SiteCommandDtoMapper.FromRouteRequest(request);
+ var reply = await DispatchAsync(command, context).ConfigureAwait(false);
+ return SiteCommandDtoMapper.ToRouteReply(reply);
+ }
+
+ ///
+ public override Task TriggerFailover(TriggerSiteFailoverDto request, ServerCallContext context)
+ {
+ var dispatcher = EnsureReady();
+ var msg = SiteCommandDtoMapper.FromProto(request);
+
+ // Resolve the standby and build the ack WITHOUT leaving. The real leave is deferred so the
+ // ack is on the wire first (ack-before-Leave).
+ var outcome = dispatcher.PrepareFailover(msg);
+ var dto = SiteCommandDtoMapper.ToProto(outcome.Ack);
+
+ if (outcome.CommitLeave is not null)
+ {
+ _leaveScheduler(outcome.CommitLeave);
+ }
+
+ return Task.FromResult(dto);
+ }
+
+ ///
+ /// Routes a decoded command through the shared dispatcher and returns the reply record.
+ /// Forward → local Ask; fire-and-forget → local Tell + synthetic ack; no target →
+ /// the dispatcher's synthetic "handler not available" reply.
+ ///
+ private async Task