dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(comm): extract SiteCommandDispatcher; site serves commands over gRPC too (T1B.2)

Browse Source 
Refactor SiteCommunicationActor's central→site routing table into one
SiteCommandDispatcher — the single routing truth for the 28 migrated commands
(IntegrationCallRequest, the dead 29th, stays on the actor and out of the
dispatcher). The Akka actor and the new SiteCommandGrpcService both route through
one dispatcher instance so the two transports can never drift on where a command
goes. Server-side only: nothing central flips to gRPC yet (that is T1B.3);
ClusterClient remains the live path.

Decisions worth recording:

- Targets preserved byte-for-byte. Lifecycle/OPC UA/query/route → the Deployment
  Manager singleton proxy; DeployArtifacts/EventLog/parked → their null-guarded
  handlers with the exact same "handler not available" replies; the parked
  handler stays NODE-LOCAL (per-node replicated-store owner), never the singleton
  proxy — pinned by a dispatcher test that asserts the target is the parked probe
  and NOT the dm proxy.

- Sender preservation intact. The actor's command handlers became thin
  DispatchCommand delegations that still Forward (central Ask → reply routes
  straight back); the existing SiteCommunicationActorTests pass unchanged, which
  is the regression guard for that plumbing. UnsubscribeDebugView keeps its
  fire-and-forget shape: the actor Forwards, the gRPC service Tells + returns the
  synthetic UnsubscribeDebugViewAck so a unary RPC still answers.

- Ack-before-Leave on failover. The dispatcher's PrepareFailover resolves the
  standby with a DRY-RUN (no leave) to build the ack, and hands back a deferred
  CommitLeave; the gRPC service returns the ack, then schedules the real
  Cluster.Leave — so a caller reaching the very node about to leave still gets its
  ack instead of a broken stream. The actor path keeps today's coupled
  resolve-and-leave (over ClusterClient the ack Tell only enqueues, so order is
  immaterial). Proven at both levels: a dispatcher test asserts the ack is built
  before CommitLeave runs, and a TestServer test asserts the recorded seam order
  is resolve-then-leave.

- ControlPlaneAuthInterceptor gates SiteCommandService by EXTENDING
  DefaultGatedPrefixes (descriptor-derived), not by adding a constructor — the
  one-public-ctor invariant and its test stay green.

Tests: SiteCommandDispatcherTests (28-command routing incl. parked node-locality
and both failover paths) and SiteCommandGrpcService TestServer tests (auth,
readiness→Unavailable, one command per oneof group, failover ordering). Full
solution build 0/0; Communication.Tests 574 and Host.Tests 377 green. No active
<Protobuf> item.
This commit is contained in:
Joseph Doherty
2026-07-22 19:10:44 -04:00
parent 59b13d317b
commit 518c699b90
9 changed files with 1315 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs
+99 -236
View File
@@ -36,13 +36,20 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
/// do not need a real cluster.
/// </summary>
private readonly Func<bool> _isActiveCheck;
private readonly Func<string, string?> _failOverRole;
/// <summary>
/// Reference to the local Deployment Manager singleton proxy.
/// </summary>
private readonly IActorRef _deploymentManagerProxy;
/// <summary>
/// The single routing truth for the 28 migrated central→site commands, shared with
/// the gRPC <c>SiteCommandGrpcService</c> 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.
/// </summary>
private readonly SiteCommandDispatcher _dispatcher;
/// <summary>
/// The site→central transport. Finalized in <see cref="PreStart"/> to the injected instance,
/// or a default <see cref="AkkaCentralTransport"/> (ClusterClient) when none is supplied — so
@@ -54,13 +61,10 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
private readonly ICentralTransport? _injectedTransport;
/// <summary>
/// Local actor references for routing specific message patterns.
/// Populated via registration messages.
/// Handler for the vestigial <see cref="IntegrationCallRequest"/> — the one command NOT
/// migrated to the dispatcher (dead at both ends), so it still routes on the actor.
/// </summary>
private IActorRef? _eventLogHandler;
private IActorRef? _parkedMessageHandler;
private IActorRef? _integrationHandler;
private IActorRef? _artifactHandler;
/// <summary>Akka timer scheduler injected by the framework via <see cref="IWithTimers"/>.</summary>
public ITimerScheduler Timers { get; set; } = null!;
@@ -82,24 +86,41 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
/// Host injects a <see cref="Grpc.GrpcCentralTransport"/> when
/// <c>ScadaBridge:Communication:CentralTransport</c> is <c>Grpc</c>.
/// </param>
/// <param name="dispatcher">
/// The shared <see cref="SiteCommandDispatcher"/> (production: created by the Host and also
/// handed to the gRPC command service, so both transports route through one instance).
/// <c>null</c> makes the actor build its own from <paramref name="failOverRole"/> — the shape
/// the existing tests use.
/// </param>
public SiteCommunicationActor(
string siteId,
CommunicationOptions options,
IActorRef deploymentManagerProxy,
Func<bool>? isActiveCheck = null,
Func<string, string?>? 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<string, bool, string?> 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<RegisterLocalHandler>(HandleRegisterLocalHandler);
// Pattern 1: Instance Deployment — forward to Deployment Manager
Receive<RefreshDeploymentCommand>(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<RefreshDeploymentCommand>(cmd => DispatchCommand(cmd));
Receive<DisableInstanceCommand>(cmd => DispatchCommand(cmd));
Receive<EnableInstanceCommand>(cmd => DispatchCommand(cmd));
Receive<DeleteInstanceCommand>(cmd => DispatchCommand(cmd));
Receive<DeploymentStateQueryRequest>(cmd => DispatchCommand(cmd));
Receive<DeployArtifactsCommand>(cmd => DispatchCommand(cmd));
Receive<SubscribeDebugViewRequest>(cmd => DispatchCommand(cmd));
Receive<UnsubscribeDebugViewRequest>(cmd => DispatchCommand(cmd));
Receive<DebugSnapshotRequest>(cmd => DispatchCommand(cmd));
Receive<RouteToCallRequest>(cmd => DispatchCommand(cmd));
Receive<RouteToGetAttributesRequest>(cmd => DispatchCommand(cmd));
Receive<RouteToSetAttributesRequest>(cmd => DispatchCommand(cmd));
Receive<RouteToWaitForAttributeRequest>(cmd => DispatchCommand(cmd));
Receive<BrowseNodeCommand>(cmd => DispatchCommand(cmd));
Receive<ReadTagValuesCommand>(cmd => DispatchCommand(cmd));
Receive<SearchAddressSpaceCommand>(cmd => DispatchCommand(cmd));
Receive<Commons.Messages.DataConnection.WriteTagRequest>(cmd => DispatchCommand(cmd));
Receive<VerifyEndpointCommand>(cmd => DispatchCommand(cmd));
Receive<TrustServerCertCommand>(cmd => DispatchCommand(cmd));
Receive<ListServerCertsCommand>(cmd => DispatchCommand(cmd));
Receive<RemoveServerCertCommand>(cmd => DispatchCommand(cmd));
Receive<EventLogQueryRequest>(cmd => DispatchCommand(cmd));
Receive<ParkedMessageQueryRequest>(cmd => DispatchCommand(cmd));
Receive<ParkedMessageRetryRequest>(cmd => DispatchCommand(cmd));
Receive<ParkedMessageDiscardRequest>(cmd => DispatchCommand(cmd));
Receive<RetryParkedOperation>(cmd => DispatchCommand(cmd));
Receive<DiscardParkedOperation>(cmd => DispatchCommand(cmd));
// Pattern 2: Lifecycle — forward to Deployment Manager
Receive<DisableInstanceCommand>(msg => _deploymentManagerProxy.Forward(msg));
Receive<EnableInstanceCommand>(msg => _deploymentManagerProxy.Forward(msg));
Receive<DeleteInstanceCommand>(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<DeploymentStateQueryRequest>(msg => _deploymentManagerProxy.Forward(msg));
// Pattern 3: Artifact Deployment — forward to artifact handler if registered
Receive<DeployArtifactsCommand>(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<IntegrationCallRequest>(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<SubscribeDebugViewRequest>(msg => _deploymentManagerProxy.Forward(msg));
Receive<UnsubscribeDebugViewRequest>(msg => _deploymentManagerProxy.Forward(msg));
// Pattern 6a: Debug Snapshot (one-shot) — forward to Deployment Manager
Receive<DebugSnapshotRequest>(msg => _deploymentManagerProxy.Forward(msg));
// Inbound API Route.To() — forward to Deployment Manager for instance routing
Receive<RouteToCallRequest>(msg => _deploymentManagerProxy.Forward(msg));
Receive<RouteToGetAttributesRequest>(msg => _deploymentManagerProxy.Forward(msg));
Receive<RouteToSetAttributesRequest>(msg => _deploymentManagerProxy.Forward(msg));
Receive<RouteToWaitForAttributeRequest>(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<BrowseNodeCommand>(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<ReadTagValuesCommand>(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<SearchAddressSpaceCommand>(msg => _deploymentManagerProxy.Forward(msg));
Receive<Commons.Messages.DataConnection.WriteTagRequest>(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<VerifyEndpointCommand>(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<TrustServerCertCommand>(msg => _deploymentManagerProxy.Forward(msg));
Receive<ListServerCertsCommand>(msg => _deploymentManagerProxy.Forward(msg));
Receive<RemoveServerCertCommand>(msg => _deploymentManagerProxy.Forward(msg));
// Pattern 7: Remote Queries
Receive<EventLogQueryRequest>(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<ParkedMessageQueryRequest>(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<ParkedMessageRetryRequest>(msg =>
{
if (_parkedMessageHandler != null)
_parkedMessageHandler.Forward(msg);
else
{
Sender.Tell(new ParkedMessageRetryResponse(
msg.CorrelationId, false, "Parked message handler not available"));
}
});
Receive<ParkedMessageDiscardRequest>(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<RetryParkedOperation>(msg =>
{
if (_parkedMessageHandler != null)
_parkedMessageHandler.Forward(msg);
else
{
Sender.Tell(new ParkedOperationActionAck(
msg.CorrelationId, Applied: false, "Parked message handler not available"));
}
});
Receive<DiscardParkedOperation>(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<TriggerSiteFailover>(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<TriggerSiteFailover>(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);
}
/// <summary>
/// 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.
/// </summary>
/// <param name="command">The migrated central→site command to route.</param>
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));
/// <summary>
/// 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.
/// </summary>
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));
}
}
/// <summary>
/// Production failover action: gracefully Leave the oldest Up member carrying
/// <paramref name="role"/>, via the shared <see cref="ClusterState.ClusterFailoverCoordinator"/>
/// so the central and site paths cannot drift. Injected in tests for the same reason
/// <see cref="DefaultIsActiveCheck"/> is — a real Leave needs Akka.Cluster in the
/// ActorSystem, which the TestKit system does not load.
/// </summary>
/// <param name="role">Site-specific role scope.</param>
/// <returns>Address of the leaving node, or null when there is no peer.</returns>
private string? DefaultFailOverRole(string role) =>
ClusterState.ClusterFailoverCoordinator.FailOverOldest(Context.System, role)?.ToString();
// ── Internal messages ──
internal record SendHeartbeat;