Joseph Doherty
674 lines
31 KiB
C#
674 lines
31 KiB
C#
using Akka.Actor;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.Event;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
|
|
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication.Actors;
|
|
|
|
/// <summary>
|
|
/// Central-side actor that routes messages from central to site clusters over the gRPC
|
|
/// <c>SiteCommandService</c> command plane (<see cref="Grpc.GrpcSiteTransport"/>). Resolves site
|
|
/// addresses from the database on a periodic refresh cycle and reconciles the transport's per-site
|
|
/// channel pairs.
|
|
///
|
|
/// All 8 message patterns routed through this actor.
|
|
/// Ask timeout on connection drop (no central buffering). Debug streams killed on interruption.
|
|
/// </summary>
|
|
public class CentralCommunicationActor : ReceiveActor
|
|
{
|
|
private readonly ILoggingAdapter _log = Context.GetLogger();
|
|
private readonly IServiceProvider _serviceProvider;
|
|
|
|
/// <summary>
|
|
/// The central→site command transport (the gRPC <see cref="Grpc.GrpcSiteTransport"/>, built by
|
|
/// the Host and injected). The <see cref="SiteEnvelope"/> handler delegates every send here, and
|
|
/// each DB refresh tick reconciles its per-site channel pairs. Assigned in the constructor body
|
|
/// before any message can arrive.
|
|
/// </summary>
|
|
private ISiteCommandTransport _transport = null!;
|
|
|
|
// The previous _debugSubscriptions / _inProgressDeployments
|
|
// dictionaries existed solely to support a documented "synchronous kill streams +
|
|
// mark deployments failed on site disconnect" workflow triggered by
|
|
// ConnectionStateChanged. No production code ever emitted that message — only
|
|
// the unit test did — so the workflow was dead from end to end. Disconnect
|
|
// detection is owned by the underlying transports: the gRPC keepalive PING
|
|
// signals stream interruption in ~25s (handled by DebugStreamBridgeActor's own
|
|
// reconnection logic), and an Ask round-trip for a deploy times out at the
|
|
// CommunicationService layer (caller sees failure). The tracking dicts +
|
|
// ConnectionStateChanged record + HandleConnectionStateChanged handler are
|
|
// removed; see docs/requirements/Component-Communication.md "Connection
|
|
// Failure Behavior" for the keepalive-based contract that survives.
|
|
|
|
private ICancelable? _refreshSchedule;
|
|
|
|
/// <summary>
|
|
/// Per-actor lifecycle CTS threaded into the periodic
|
|
/// <see cref="LoadSiteAddressesFromDb"/> repository call so a hung MS SQL
|
|
/// connection is bounded by actor shutdown rather than holding piped tasks
|
|
/// open indefinitely. Cancelled in <see cref="PostStop"/>; never reset.
|
|
/// </summary>
|
|
private readonly CancellationTokenSource _lifecycleCts = new();
|
|
|
|
/// <summary>
|
|
/// Process-wide monotonic stamp for <see cref="SiteAddressCacheLoaded"/> (review 2026-10-21
|
|
/// 03/M-1). Static so it stays monotonic across actor restarts: a late result piped from a
|
|
/// previous incarnation's load can never outrank a newer one.
|
|
/// </summary>
|
|
private static long s_loadSequence;
|
|
|
|
/// <summary>Sequence of the DB load currently in flight; 0 when none (the overlap guard).</summary>
|
|
private long _inFlightLoadSequence;
|
|
|
|
/// <summary>A refresh requested while a load was in flight; re-run once the load completes.</summary>
|
|
private bool _refreshPending;
|
|
|
|
/// <summary>Sequence of the last result handed to the transport; older results are dropped.</summary>
|
|
private long _lastAppliedLoadSequence;
|
|
|
|
/// <summary>
|
|
/// Proxy <see cref="IActorRef"/> for the central NotificationOutboxActor cluster singleton.
|
|
/// Set via <see cref="RegisterNotificationOutbox"/> — the Host creates the singleton proxy
|
|
/// after this actor and registers it (mirrors how the site-side actor receives its
|
|
/// runtime <see cref="IActorRef"/>s). Null until registration completes; a notification
|
|
/// arriving before then is rejected with a non-accepted ack so the site retries.
|
|
/// </summary>
|
|
private IActorRef? _notificationOutboxProxy;
|
|
|
|
/// <summary>
|
|
/// DistributedPubSub topic used to fan health reports out to the peer
|
|
/// central node so both per-node aggregators stay in sync. See
|
|
/// <see cref="SiteHealthReportReplica"/> for the protocol rationale.
|
|
/// </summary>
|
|
private const string HealthReportTopic = "site-health-replica";
|
|
|
|
/// <summary>
|
|
/// Constructs the actor over the given central→site command transport (production injects the
|
|
/// gRPC <see cref="Grpc.GrpcSiteTransport"/>, built by the Host from the DI
|
|
/// <see cref="Grpc.SitePairChannelProvider"/>; tests substitute a fake).
|
|
/// </summary>
|
|
/// <param name="serviceProvider">DI service provider for scoped repository and aggregator access.</param>
|
|
/// <param name="transport">The central→site command transport to route every <see cref="SiteEnvelope"/> through.</param>
|
|
public CentralCommunicationActor(
|
|
IServiceProvider serviceProvider,
|
|
ISiteCommandTransport transport)
|
|
: this(serviceProvider)
|
|
{
|
|
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
|
|
}
|
|
|
|
/// <summary>Shared wiring: sets scoped state and registers every message handler. The
|
|
/// <see cref="_transport"/> is assigned by the delegating public constructor before any message
|
|
/// is dispatched.</summary>
|
|
/// <param name="serviceProvider">DI service provider.</param>
|
|
private CentralCommunicationActor(IServiceProvider serviceProvider)
|
|
{
|
|
_serviceProvider = serviceProvider;
|
|
|
|
// Site address cache loaded from database
|
|
Receive<SiteAddressCacheLoaded>(HandleSiteAddressCacheLoaded);
|
|
|
|
// Periodic refresh trigger
|
|
Receive<RefreshSiteAddresses>(_ => LoadSiteAddressesFromDb());
|
|
|
|
// A faulted (or deadline-cancelled) LoadSiteAddressesFromDb task. Without this the
|
|
// failure was an unhandled message (debug-level only) and the refresh failed silently —
|
|
// operators could not distinguish "no sites configured" from "database is down". Log at
|
|
// Warning and release the overlap guard.
|
|
Receive<SiteAddressLoadFailed>(HandleSiteAddressLoadFailed);
|
|
// The load task converts its own faults to SiteAddressLoadFailed, so a Status.Failure
|
|
// here is unexpected — but it must still release the overlap guard, or every later
|
|
// refresh would be coalesced behind a load that never reports back.
|
|
Receive<Status.Failure>(failure =>
|
|
{
|
|
_log.Warning(failure.Cause,
|
|
"Failed to load site addresses from the database; the per-site gRPC channel "
|
|
+ "cache was not refreshed and may be stale or empty");
|
|
if (_inFlightLoadSequence != 0)
|
|
{
|
|
CompleteLoad(_inFlightLoadSequence);
|
|
}
|
|
});
|
|
|
|
// Health monitoring: heartbeats and health reports from sites
|
|
Receive<HeartbeatMessage>(HandleHeartbeat);
|
|
Receive<SiteHealthReport>(HandleSiteHealthReport);
|
|
Receive<SiteHealthReportReplica>(r => ProcessLocally(r.Report));
|
|
Receive<SiteHeartbeatReplica>(r => MarkHeartbeatLocally(r.Heartbeat));
|
|
Receive<SubscribeAck>(_ => { /* DistributedPubSub subscribe confirmation */ });
|
|
|
|
// Route enveloped messages to sites
|
|
Receive<SiteEnvelope>(HandleSiteEnvelope);
|
|
|
|
// Notification Outbox: the Host registers the outbox singleton proxy after this
|
|
// actor is created (the proxy cannot exist before this actor's construction).
|
|
Receive<RegisterNotificationOutbox>(msg =>
|
|
{
|
|
_notificationOutboxProxy = msg.OutboxProxy;
|
|
_log.Info("Registered notification outbox proxy");
|
|
});
|
|
|
|
// Notification Outbox ingest: a site forwards a buffered NotificationSubmit to the
|
|
// central cluster. Forward to the outbox proxy so the original
|
|
// Sender (the site's transport path) is preserved and the NotificationSubmitAck
|
|
// routes straight back to the site.
|
|
Receive<NotificationSubmit>(HandleNotificationSubmit);
|
|
|
|
// Notification Outbox status query: forward to the outbox proxy, preserving Sender
|
|
// so the NotificationStatusResponse routes back to the querying site.
|
|
Receive<NotificationStatusQuery>(HandleNotificationStatusQuery);
|
|
|
|
// Audit Log site→central ingest is NOT relayed here. The central-hosted
|
|
// CentralControlGrpcService Asks the audit-log-ingest singleton proxy directly
|
|
// (as SiteStreamGrpcServer always has), so this actor no longer sits in that
|
|
// path — the relay was a second hop with an identical 30 s Ask timeout that
|
|
// could only add latency. Do not reintroduce it.
|
|
|
|
// Startup reconciliation: a site node forwards its local deployed inventory on
|
|
// startup. Resolve the scoped ReconcileService, diff the
|
|
// inventory against central's expected set, and pipe the ReconcileSiteResponse
|
|
// (gap fetch tokens + orphans) straight back to the site node.
|
|
Receive<ReconcileSiteRequest>(msg => HandleReconcileSiteRequest(msg, CancellationToken.None));
|
|
|
|
// The same request as received by CentralControlGrpcService.ReconcileSite, carrying the
|
|
// gRPC call's cancellation token so central stops staging work for a site that gave up.
|
|
Receive<ReconcileSiteCall>(call => HandleReconcileSiteRequest(call.Request, call.CallCancelled));
|
|
}
|
|
|
|
private void HandleNotificationSubmit(NotificationSubmit msg)
|
|
{
|
|
if (_notificationOutboxProxy == null)
|
|
{
|
|
// No outbox proxy registered yet. A non-accepted ack makes the site's
|
|
// Store-and-Forward forwarder treat this as transient and retry later.
|
|
_log.Warning(
|
|
"Cannot route NotificationSubmit {0} — notification outbox not available",
|
|
msg.NotificationId);
|
|
Sender.Tell(new NotificationSubmitAck(
|
|
msg.NotificationId, Accepted: false, Error: "notification outbox not available"));
|
|
return;
|
|
}
|
|
|
|
_log.Debug("Routing NotificationSubmit {0} to the notification outbox", msg.NotificationId);
|
|
_notificationOutboxProxy.Forward(msg);
|
|
}
|
|
|
|
private void HandleNotificationStatusQuery(NotificationStatusQuery msg)
|
|
{
|
|
if (_notificationOutboxProxy == null)
|
|
{
|
|
// No outbox proxy registered yet. Reply Found: false so the querying site
|
|
// falls back to its local Store-and-Forward buffer to resolve the status.
|
|
_log.Warning(
|
|
"Cannot route NotificationStatusQuery {0} — notification outbox not available",
|
|
msg.NotificationId);
|
|
Sender.Tell(new NotificationStatusResponse(
|
|
msg.CorrelationId, Found: false, Status: "Unknown",
|
|
RetryCount: 0, LastError: null, DeliveredAt: null));
|
|
return;
|
|
}
|
|
|
|
_log.Debug("Routing NotificationStatusQuery {0} to the notification outbox", msg.NotificationId);
|
|
_notificationOutboxProxy.Forward(msg);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Startup reconciliation (site→central): resolve the scoped
|
|
/// <see cref="ReconcileService"/> in a DI scope, diff the node's reported inventory
|
|
/// against central's expected set, and pipe the <see cref="ReconcileSiteResponse"/>
|
|
/// back to the site node's transport path. The actor stays thin — all the diff
|
|
/// and staging logic lives in the service. Mirrors the DB-access pattern used by
|
|
/// <see cref="LoadSiteAddressesFromDb"/> (Task.Run + CreateScope + PipeTo) and the
|
|
/// Sender-preservation pattern of <see cref="HandleNotificationSubmit"/>.
|
|
///
|
|
/// On a faulted task PipeTo delivers a <see cref="Status.Failure"/> to the node; its
|
|
/// Ask faults and it simply retries reconcile on the next startup — reconcile is
|
|
/// best-effort, so the fault is allowed to propagate rather than being swallowed.
|
|
/// </summary>
|
|
/// <param name="msg">The site node's reconcile request.</param>
|
|
/// <param name="callCancelled">
|
|
/// The originating gRPC call's token (review 2026-10-21 03/P-1b): when the site's deadline
|
|
/// expires or it disconnects, the reconcile stops instead of staging pending rows for a caller
|
|
/// that is gone. <see cref="CancellationToken.None"/> for an in-process caller.
|
|
/// </param>
|
|
private void HandleReconcileSiteRequest(ReconcileSiteRequest msg, CancellationToken callCancelled)
|
|
{
|
|
// Capture Sender before the async/PipeTo — Akka resets Sender between dispatches.
|
|
var replyTo = Sender;
|
|
|
|
// Bound the DB work by the actor lifecycle. The CTS may have
|
|
// been disposed by PostStop on a racing late message; treat that as "actor gone".
|
|
CancellationToken ct;
|
|
try
|
|
{
|
|
ct = _lifecycleCts.Token;
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_log.Debug(
|
|
"Handling ReconcileSiteRequest from site {0} node {1} ({2} local instance(s))",
|
|
msg.SiteIdentifier, msg.NodeId, msg.LocalNameToRevisionHash.Count);
|
|
|
|
Task.Run(async () =>
|
|
{
|
|
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, callCancelled);
|
|
using var scope = _serviceProvider.CreateScope();
|
|
var service = scope.ServiceProvider.GetRequiredService<ReconcileService>();
|
|
return await service.ReconcileAsync(msg, linked.Token).ConfigureAwait(false);
|
|
}).PipeTo(replyTo);
|
|
}
|
|
|
|
private void HandleHeartbeat(HeartbeatMessage heartbeat)
|
|
{
|
|
// Synthetic heartbeats carry no liveness meaning — today only
|
|
// CentralChannelProvider's failback probe, which reuses this RPC to ask whether the
|
|
// preferred central endpoint answers again. It is emitted by a site's TRANSPORT layer,
|
|
// not by any node's heartbeat timer, so stamping it would let a site whose real
|
|
// heartbeats had stopped keep looking alive on the health dashboard for as long as its
|
|
// transport kept probing. Dropped before the local mark AND before the peer fan-out,
|
|
// so neither central node's aggregator ever sees it.
|
|
if (heartbeat.Synthetic)
|
|
{
|
|
_log.Debug(
|
|
"Ignoring synthetic heartbeat from site {0} (node '{1}') — probe traffic, not liveness",
|
|
heartbeat.SiteId, heartbeat.NodeHostname);
|
|
return;
|
|
}
|
|
|
|
MarkHeartbeatLocally(heartbeat);
|
|
|
|
// Fan the heartbeat out to the peer central node so BOTH aggregators mark
|
|
// it, regardless of which central node the site delivered
|
|
// to. Without this, a heartbeat that only ever reaches one node leaves the
|
|
// other node's aggregator blind to that site's liveness after a failover
|
|
// (arch review 02, Low). MarkHeartbeat is idempotent (timestamp overwrite),
|
|
// so self-delivery via pub-sub is harmless.
|
|
try
|
|
{
|
|
DistributedPubSub.Get(Context.System).Mediator.Tell(
|
|
new Publish(HealthReportTopic, new SiteHeartbeatReplica(heartbeat)));
|
|
}
|
|
catch
|
|
{
|
|
// No-op in non-clustered hosts (TestKit).
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Marks a site heartbeat on the local aggregator without re-broadcasting.
|
|
/// Used for both site-originated heartbeats and peer-replicated ones.
|
|
/// A synthetic heartbeat is never marked — <see cref="HandleHeartbeat"/> already drops it
|
|
/// before the fan-out, so a replica can only carry one if a peer node predates the flag;
|
|
/// the guard here is belt-and-braces on the last hop before the aggregator.
|
|
/// </summary>
|
|
private void MarkHeartbeatLocally(HeartbeatMessage heartbeat)
|
|
{
|
|
if (heartbeat.Synthetic)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var aggregator = _serviceProvider.GetService<ICentralHealthAggregator>();
|
|
aggregator?.MarkHeartbeat(heartbeat.SiteId, heartbeat.Timestamp);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles a report delivered directly from a site:
|
|
/// process locally, then fan out to the peer central node so its
|
|
/// aggregator stays in sync.
|
|
/// </summary>
|
|
private void HandleSiteHealthReport(SiteHealthReport report)
|
|
{
|
|
ProcessLocally(report);
|
|
|
|
try
|
|
{
|
|
DistributedPubSub.Get(Context.System).Mediator.Tell(
|
|
new Publish(HealthReportTopic, new SiteHealthReportReplica(report)));
|
|
}
|
|
catch
|
|
{
|
|
// No-op in non-clustered hosts (TestKit).
|
|
}
|
|
|
|
// Ack the site so its AkkaHealthReportTransport Ask completes and the
|
|
// report-loss counter-restore path can observe delivery (review 01
|
|
// [Medium]). Guarded so the peer-replica path (SiteHealthReportReplica,
|
|
// which arrives without an Ask sender) never dead-letters an ack.
|
|
if (!Sender.IsNobody())
|
|
{
|
|
Sender.Tell(new SiteHealthReportAck(report.SiteId, report.SequenceNumber, Accepted: true));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Applies a report to the local aggregator without re-broadcasting.
|
|
/// Used for both site-originated reports and peer-replicated ones — the
|
|
/// aggregator is idempotent via sequence-number comparison.
|
|
/// </summary>
|
|
private void ProcessLocally(SiteHealthReport report)
|
|
{
|
|
var aggregator = _serviceProvider.GetService<ICentralHealthAggregator>();
|
|
if (aggregator != null)
|
|
{
|
|
aggregator.ProcessReport(report);
|
|
}
|
|
else
|
|
{
|
|
_log.Warning("ICentralHealthAggregator not available, dropping health report from site {0}", report.SiteId);
|
|
}
|
|
}
|
|
|
|
// HandleConnectionStateChanged removed — no production
|
|
// caller emitted ConnectionStateChanged, so the workflow ran only in tests.
|
|
// Disconnect detection is owned by the transport layers (gRPC keepalive +
|
|
// Ask timeout).
|
|
|
|
private void HandleSiteEnvelope(SiteEnvelope envelope)
|
|
{
|
|
// Below-the-seam routing: the gRPC transport owns the "no route for this site ⇒ warn +
|
|
// drop, caller's Ask times out" contract. Sender is the temporary Ask actor (or the
|
|
// debug-bridge actor) and is preserved for reply routing.
|
|
_transport.Send(envelope, Sender);
|
|
}
|
|
|
|
private void LoadSiteAddressesFromDb()
|
|
{
|
|
// Overlap guard (review 2026-10-21 03/M-1): a wedged DB used to stack one Task.Run + DI
|
|
// scope + DbContext per 60 s tick, and their late completions could land out of order.
|
|
// A tick (or an admin-change refresh) that arrives mid-load is coalesced into ONE re-run
|
|
// once the current load finishes, so an on-demand refresh is never lost.
|
|
if (_inFlightLoadSequence != 0)
|
|
{
|
|
_refreshPending = true;
|
|
return;
|
|
}
|
|
|
|
var self = Self;
|
|
// Pass the actor's lifecycle CT into the repository
|
|
// call so a hung database query is cancelled when the actor stops
|
|
// rather than leaving the piped task to accumulate. Captured locally
|
|
// because the lifecycle CTS may have been disposed by PostStop on a
|
|
// racing late tick; treat that as "actor gone, give up".
|
|
CancellationToken lifecycle;
|
|
try
|
|
{
|
|
lifecycle = _lifecycleCts.Token;
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var sequence = Interlocked.Increment(ref s_loadSequence);
|
|
_inFlightLoadSequence = sequence;
|
|
_refreshPending = false;
|
|
// Per-load deadline: a hung read is abandoned instead of pinning the overlap guard.
|
|
var deadline = _serviceProvider.GetService<IOptions<CommunicationOptions>>()?.Value.QueryTimeout
|
|
?? TimeSpan.FromSeconds(30);
|
|
|
|
Task.Run(async () =>
|
|
{
|
|
CancellationTokenSource? loadCts = null;
|
|
try
|
|
{
|
|
// Inside the try: a throw here (e.g. a lifecycle CTS disposed by a racing
|
|
// PostStop) must still report back, or the overlap guard would never release.
|
|
loadCts = CancellationTokenSource.CreateLinkedTokenSource(lifecycle);
|
|
loadCts.CancelAfter(deadline);
|
|
using var scope = _serviceProvider.CreateScope();
|
|
var repo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
|
|
var sites = await repo.GetAllSitesAsync(loadCts.Token).ConfigureAwait(false);
|
|
|
|
// gRPC-endpoint cache (the GrpcNodeA/BAddress columns, NOT the legacy Akka
|
|
// NodeA/BAddress ones, which no longer drive cross-cluster dialling).
|
|
var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
|
|
foreach (var site in sites)
|
|
{
|
|
var grpcA = string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) ? null : site.GrpcNodeAAddress;
|
|
var grpcB = string.IsNullOrWhiteSpace(site.GrpcNodeBAddress) ? null : site.GrpcNodeBAddress;
|
|
if (grpcA is not null || grpcB is not null)
|
|
grpcContacts[site.SiteIdentifier] = new SiteGrpcEndpoints(grpcA, grpcB);
|
|
}
|
|
|
|
// Carry the full configured-site identifier set (not just the
|
|
// gRPC-address-bearing subset) so the aggregator prunes only
|
|
// genuinely-deleted sites and never an addressless-but-configured one.
|
|
var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
|
|
return (object)new SiteAddressCacheLoaded(knownSiteIds, grpcContacts, sequence);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
var stopping = lifecycle.IsCancellationRequested;
|
|
var timedOut = ex is OperationCanceledException && !stopping;
|
|
return new SiteAddressLoadFailed(sequence, ex, timedOut ? deadline : null, ActorStopping: stopping);
|
|
}
|
|
finally
|
|
{
|
|
loadCts?.Dispose();
|
|
}
|
|
}).PipeTo(self);
|
|
}
|
|
|
|
private void CompleteLoad(long sequence)
|
|
{
|
|
if (sequence != _inFlightLoadSequence)
|
|
{
|
|
return; // a previous incarnation's / superseded load — the guard belongs to another one
|
|
}
|
|
_inFlightLoadSequence = 0;
|
|
if (_refreshPending)
|
|
{
|
|
LoadSiteAddressesFromDb();
|
|
}
|
|
}
|
|
|
|
private void HandleSiteAddressLoadFailed(SiteAddressLoadFailed failed)
|
|
{
|
|
if (failed.ActorStopping)
|
|
{
|
|
// Cancelled because the (previous incarnation of the) actor stopped or restarted —
|
|
// expected, not a database problem.
|
|
_log.Debug("Site-address load {0} cancelled by actor shutdown/restart", failed.Sequence);
|
|
}
|
|
else if (failed.TimedOutAfter is { } after)
|
|
{
|
|
_log.Warning(
|
|
"Loading site addresses from the database did not complete within {0}; the per-site "
|
|
+ "gRPC channel cache was not refreshed and may be stale or empty", after);
|
|
}
|
|
else
|
|
{
|
|
_log.Warning(failed.Cause,
|
|
"Failed to load site addresses from the database; the per-site gRPC channel "
|
|
+ "cache was not refreshed and may be stale or empty");
|
|
}
|
|
CompleteLoad(failed.Sequence);
|
|
}
|
|
|
|
private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg)
|
|
{
|
|
CompleteLoad(msg.Sequence);
|
|
|
|
// Never let an older read overwrite a newer one (review 2026-10-21 03/M-1): applying
|
|
// stale addresses would park freshly built channels and rebuild from the old endpoints.
|
|
if (msg.Sequence <= _lastAppliedLoadSequence)
|
|
{
|
|
_log.Debug("Dropping stale site-address load {0} (last applied {1})",
|
|
msg.Sequence, _lastAppliedLoadSequence);
|
|
return;
|
|
}
|
|
_lastAppliedLoadSequence = msg.Sequence;
|
|
|
|
// Per-site transport resource reconciliation (build/drop gRPC channel
|
|
// pairs). Runs on the actor thread each refresh tick. Guarded (review 2026-10-21 03/R-1):
|
|
// a throw here would restart this actor, whose PreStart re-reads the Sites table at
|
|
// TimeSpan.Zero — a restart loop for as long as a bad row persists — and would skip the
|
|
// aggregator prune below.
|
|
try
|
|
{
|
|
_transport.ReconcileSites(msg);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "Reconciling site gRPC channels from the refreshed address cache failed; "
|
|
+ "existing channels are kept until the next refresh");
|
|
}
|
|
|
|
// Self-healing eviction: a site deleted from configuration would otherwise
|
|
// linger in the aggregator as a permanently-offline tile (and live KPI
|
|
// sample source) forever. Prune on every refresh so it disappears within
|
|
// one refresh interval without needing a dedicated deletion event. Transport-agnostic.
|
|
_serviceProvider.GetService<ICentralHealthAggregator>()?.PruneUnknownSites(msg.KnownSiteIds);
|
|
}
|
|
|
|
// TrackMessageForCleanup removed — the dicts it fed
|
|
// existed solely to support the dead ConnectionStateChanged workflow.
|
|
|
|
/// <inheritdoc />
|
|
protected override SupervisorStrategy SupervisorStrategy()
|
|
{
|
|
return new OneForOneStrategy(
|
|
maxNrOfRetries: -1,
|
|
withinTimeRange: Timeout.InfiniteTimeSpan,
|
|
decider: Decider.From(ex =>
|
|
{
|
|
_log.Warning(ex, "Child actor of CentralCommunicationActor faulted, resuming (state preserved)");
|
|
return Directive.Resume;
|
|
}));
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override void PreStart()
|
|
{
|
|
_log.Info("CentralCommunicationActor started");
|
|
|
|
// Subscribe to the peer-replication topic so we receive health reports
|
|
// delivered to the other central node and keep our local aggregator
|
|
// in sync (a site may deliver its report to either central node).
|
|
// Tolerant of non-clustered hosts (TestKit) where the extension is absent.
|
|
try
|
|
{
|
|
DistributedPubSub.Get(Context.System).Mediator.Tell(
|
|
new Subscribe(HealthReportTopic, Self));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Debug("DistributedPubSub not available — peer health replication disabled: {0}", ex.Message);
|
|
}
|
|
|
|
// Schedule periodic refresh of site addresses from the database
|
|
_refreshSchedule = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
|
|
TimeSpan.Zero,
|
|
TimeSpan.FromSeconds(60),
|
|
Self,
|
|
new RefreshSiteAddresses(),
|
|
ActorRefs.NoSender);
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override void PostStop()
|
|
{
|
|
_log.Info("CentralCommunicationActor stopped");
|
|
_refreshSchedule?.Cancel();
|
|
// Cancel any in-flight LoadSiteAddressesFromDb so a
|
|
// hung MS SQL query does not outlive the actor.
|
|
try
|
|
{
|
|
_lifecycleCts.Cancel();
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
// Double-stop is benign.
|
|
}
|
|
_lifecycleCts.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Command to trigger a refresh of site addresses from the database.
|
|
/// </summary>
|
|
public record RefreshSiteAddresses;
|
|
|
|
/// <summary>
|
|
/// Message carrying the loaded site contact data from the database. Per-transport resource
|
|
/// reconciliation happens on the actor thread in HandleSiteAddressCacheLoaded, which hands this
|
|
/// straight to <see cref="ISiteCommandTransport.ReconcileSites"/>. In-process only (never on the
|
|
/// wire); the payload is exposed through read-only collection types so the Akka.NET "messages are
|
|
/// immutable" convention is enforced at the type level.
|
|
/// </summary>
|
|
/// <param name="KnownSiteIds">Every configured site id, address-bearing or not, for aggregator pruning.</param>
|
|
/// <param name="GrpcContacts">gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress).</param>
|
|
/// <param name="Sequence">
|
|
/// Monotonic load stamp; the actor drops a result whose sequence is not newer than the last one it
|
|
/// applied, so an out-of-order completion never re-applies stale addresses.
|
|
/// </param>
|
|
public sealed record SiteAddressCacheLoaded(
|
|
IReadOnlyCollection<string> KnownSiteIds,
|
|
IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts,
|
|
long Sequence);
|
|
|
|
/// <summary>
|
|
/// In-process envelope (never on the wire) pairing a site's <see cref="ReconcileSiteRequest"/>
|
|
/// with the cancellation token of the gRPC call that carried it, so the central reconcile stops
|
|
/// when the site abandons the call (review 2026-10-21 03/P-1b).
|
|
/// </summary>
|
|
/// <param name="Request">The reconcile request.</param>
|
|
/// <param name="CallCancelled">The gRPC call's <c>ServerCallContext.CancellationToken</c>.</param>
|
|
internal sealed record ReconcileSiteCall(ReconcileSiteRequest Request, CancellationToken CallCancelled);
|
|
|
|
/// <summary>A failed (or deadline-cancelled) site-address DB load, piped back to release the overlap guard.</summary>
|
|
/// <param name="Sequence">The failed load's sequence stamp.</param>
|
|
/// <param name="Cause">The failure.</param>
|
|
/// <param name="TimedOutAfter">The per-load deadline when the load was cancelled by it; otherwise null.</param>
|
|
/// <param name="ActorStopping">True when the actor's lifecycle token was cancelled (stop/restart) — logged at Debug.</param>
|
|
internal sealed record SiteAddressLoadFailed(long Sequence, Exception Cause, TimeSpan? TimedOutAfter, bool ActorStopping = false);
|
|
|
|
/// <summary>
|
|
/// A site's gRPC node-pair endpoints, as loaded from <c>Site.GrpcNodeAAddress</c>/
|
|
/// <c>GrpcNodeBAddress</c>. Either may be null when only one node has a gRPC address configured.
|
|
/// </summary>
|
|
/// <param name="NodeA">NodeA gRPC base address (e.g. <c>http://scadabridge-site-a-node-a:8083</c>), or null.</param>
|
|
/// <param name="NodeB">NodeB gRPC base address, or null.</param>
|
|
public readonly record struct SiteGrpcEndpoints(string? NodeA, string? NodeB);
|
|
|
|
/// <summary>
|
|
/// Peer-replication envelope for a site heartbeat, fanned out over the same
|
|
/// DistributedPubSub topic as <see cref="SiteHealthReportReplica"/> so both
|
|
/// central aggregators mark heartbeats regardless of which node the site
|
|
/// delivered to. Only ever travels central↔central (same assembly
|
|
/// version, rolled together), so the default reflective serializer is safe.
|
|
/// </summary>
|
|
public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat);
|
|
|
|
/// <summary>
|
|
/// Notification sent to debug view subscribers when the stream is terminated
|
|
/// due to site disconnection.
|
|
/// </summary>
|
|
public record DebugStreamTerminated(string SiteId, string CorrelationId);
|
|
|
|
/// <summary>
|
|
/// Registers the central NotificationOutboxActor singleton proxy with the
|
|
/// <see cref="CentralCommunicationActor"/> so site-forwarded <see cref="NotificationSubmit"/>
|
|
/// and <see cref="NotificationStatusQuery"/> messages can be routed to it. Sent by the Host
|
|
/// after the outbox singleton proxy is created.
|
|
/// </summary>
|
|
public record RegisterNotificationOutbox(IActorRef OutboxProxy);
|
|
|
|
// NOTE: there is deliberately no RegisterAuditIngest counterpart. The audit-log-ingest
|
|
// singleton proxy goes STRAIGHT to the two gRPC servers that need it
|
|
// (CentralControlGrpcService.SetAuditIngestActor);
|
|
// routing audit batches through this actor was a redundant hop and was removed.
|