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

feat(comm): T1B.3 — central-side gRPC site-command transport seam (default Akka)

Browse Source 
Extract the central→site send path in CentralCommunicationActor behind a new
ISiteCommandTransport, selected by ScadaBridge:Communication:SiteTransport
(Akka | Grpc, default Akka — rollback = flip the flag). CommunicationService's
27 commands, SiteCallAuditActor's 2 parked relays and DebugStreamBridgeActor's
subscribe/unsubscribe are untouched; the seam sits below SiteEnvelope.

- AkkaSiteTransport: today's per-site ClusterClient path extracted verbatim
  (the _siteClients lookup + ClusterClient.Send with the reply-to sender
  preserved, and the "no client ⇒ warn + drop, caller's Ask times out" path).
- GrpcSiteTransport: dials the site SiteCommandService (T1B.1 proto client) via
  SiteCommandDtoMapper, PSK + x-scadabridge-site on the channel through
  ControlPlaneCredentials, per-command deadlines set EQUAL to today's
  CommunicationService Ask timeouts (per-command, not per-group: DeploymentState
  query and TriggerSiteFailover use QueryTimeout; the two parked relays map to
  QueryTimeout so SiteCallAudit's inner RelayTimeout 10s < 30s ordering holds;
  WaitForAttribute keeps its dynamic Timeout + IntegrationTimeout).
- SitePairChannelProvider: per-site A/B channel pair with sticky failover
  (flip only on Unavailable — NEVER on DeadlineExceeded, a write/deploy/failover
  may have run), background failback probe to the preferred node with 1s→60s
  doubling backoff, PSK invalidation on site removal. Fed by the SAME DB refresh
  loop (extended to carry GrpcNodeA/GrpcNodeBAddress) — no second poll.

Tests: actor-with-substitute-transport (routing, Ask-reply plumbing, per-site
lifecycle across refreshes), ResolveDeadline pinned to each command's current
Ask timeout, and GrpcSiteTransport/SitePairChannelProvider over dual in-process
TestServers (PSK+header+deadline attached, Unavailable failover + stickiness,
failback to preferred, no-retry-on-DeadlineExceeded). Proto csproj untouched
(no active <Protobuf> item). Full solution builds 0 warnings; Communication.Tests
607 green with Akka default.
This commit is contained in:
Joseph Doherty
2026-07-22 19:41:42 -04:00
parent 518c699b90
commit 86ad4d5c8e
12 changed files with 1474 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs
+112 -89
View File
@@ -4,6 +4,8 @@ using Akka.Cluster.Tools.Client;
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;
@@ -78,14 +80,15 @@ public class CentralCommunicationActor : ReceiveActor
{
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly IServiceProvider _serviceProvider;
private readonly ISiteClientFactory _siteClientFactory;
/// <summary>
/// Per-site ClusterClient instances and their contact addresses.
/// Maps SiteIdentifier → (ClusterClient actor, set of contact address strings).
/// Refreshed periodically via RefreshSiteAddresses.
/// The active central→site command transport, chosen by
/// <c>ScadaBridge:Communication:SiteTransport</c> (default <see cref="SiteTransportKind.Akka"/>).
/// The <see cref="SiteEnvelope"/> handler delegates every send here, and each DB refresh tick
/// reconciles its per-site resources (ClusterClients for Akka, channel pairs for gRPC). Assigned
/// in the public constructor body before any message can arrive.
/// </summary>
private readonly Dictionary<string, (IActorRef Client, ImmutableHashSet<string> ContactAddresses)> _siteClients = new();
private ISiteCommandTransport _transport = null!;
// The previous _debugSubscriptions / _inProgressDeployments
// dictionaries existed solely to support a documented "synchronous kill streams +
@@ -156,9 +159,15 @@ public class CentralCommunicationActor : ReceiveActor
/// </summary>
private const string HealthReportTopic = "site-health-replica";
/// <summary>Initializes the <see cref="CentralCommunicationActor"/> and wires all message handlers.</summary>
/// <summary>
/// Legacy constructor: builds the transport by reading
/// <c>ScadaBridge:Communication:SiteTransport</c> and wrapping <paramref name="siteClientFactory"/>
/// in an <see cref="AkkaSiteTransport"/> (default) or resolving the gRPC transport from
/// <paramref name="serviceProvider"/>. Kept so the Host and the existing TestKit suites construct
/// the actor exactly as before (the factory is still the Akka seam).
/// </summary>
/// <param name="serviceProvider">DI service provider for scoped repository and aggregator access.</param>
/// <param name="siteClientFactory">Factory used to create per-site ClusterClient actors.</param>
/// <param name="siteClientFactory">Factory used to create per-site ClusterClient actors (Akka transport).</param>
/// <param name="auditIngestAskTimeout">
/// Optional override for the audit-ingest Ask timeout; defaults to
/// <see cref="Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout"/> (30 s). Exists only so tests can
@@ -168,9 +177,38 @@ public class CentralCommunicationActor : ReceiveActor
IServiceProvider serviceProvider,
ISiteClientFactory siteClientFactory,
TimeSpan? auditIngestAskTimeout = null)
: this(serviceProvider, auditIngestAskTimeout)
{
ArgumentNullException.ThrowIfNull(siteClientFactory);
_transport = SelectTransport(serviceProvider, siteClientFactory);
}
/// <summary>
/// Primary constructor: takes the already-selected <see cref="ISiteCommandTransport"/> directly.
/// Used by tests that substitute the transport, and reachable via the legacy constructor.
/// </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>
/// <param name="auditIngestAskTimeout">Optional override for the audit-ingest Ask timeout (test hook).</param>
public CentralCommunicationActor(
IServiceProvider serviceProvider,
ISiteCommandTransport transport,
TimeSpan? auditIngestAskTimeout = null)
: this(serviceProvider, auditIngestAskTimeout)
{
_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>
/// <param name="auditIngestAskTimeout">Optional audit-ingest Ask timeout override.</param>
private CentralCommunicationActor(
IServiceProvider serviceProvider,
TimeSpan? auditIngestAskTimeout)
{
_serviceProvider = serviceProvider;
_siteClientFactory = siteClientFactory;
_auditIngestAskTimeout = auditIngestAskTimeout ?? Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout;
// Site address cache loaded from database
@@ -451,19 +489,41 @@ public class CentralCommunicationActor : ReceiveActor
private void HandleSiteEnvelope(SiteEnvelope envelope)
{
if (!_siteClients.TryGetValue(envelope.SiteId, out var entry))
{
_log.Warning("No ClusterClient for site {0}, cannot route message {1}",
envelope.SiteId, envelope.Message.GetType().Name);
// Below-the-seam routing: the active transport (Akka ClusterClient or gRPC) 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);
}
// The Ask will timeout on the caller side — no central buffering
return;
/// <summary>
/// Chooses the transport from <c>ScadaBridge:Communication:SiteTransport</c> (default
/// <see cref="SiteTransportKind.Akka"/>). For Akka, wraps the injected
/// <see cref="ISiteClientFactory"/> in an <see cref="AkkaSiteTransport"/> bound to this actor's
/// context; for gRPC, resolves the shared <see cref="Grpc.SitePairChannelProvider"/> and options.
/// Runs in the constructor body, where <see cref="ActorBase.Context"/> is available.
/// </summary>
/// <param name="serviceProvider">The DI provider carrying options and (for gRPC) the channel provider.</param>
/// <param name="siteClientFactory">The ClusterClient factory used by the Akka transport.</param>
/// <returns>The selected transport.</returns>
private ISiteCommandTransport SelectTransport(
IServiceProvider serviceProvider, ISiteClientFactory siteClientFactory)
{
var options = serviceProvider.GetService<IOptions<CommunicationOptions>>()?.Value;
var kind = options?.SiteTransport ?? SiteTransportKind.Akka;
if (kind == SiteTransportKind.Grpc)
{
var channelProvider = serviceProvider.GetRequiredService<Grpc.SitePairChannelProvider>();
var loggerFactory = serviceProvider.GetRequiredService<ILoggerFactory>();
_log.Info("central→site command transport: gRPC (SiteCommandService)");
return new Grpc.GrpcSiteTransport(
channelProvider,
options!,
loggerFactory.CreateLogger<Grpc.GrpcSiteTransport>());
}
// Route via ClusterClient — Sender is preserved for Ask response routing
entry.Client.Tell(
new ClusterClient.Send("/user/site-communication", envelope.Message),
Sender);
_log.Info("central→site command transport: Akka ClusterClient");
return new AkkaSiteTransport(siteClientFactory, Context, _log);
}
private void LoadSiteAddressesFromDb()
@@ -491,6 +551,10 @@ public class CentralCommunicationActor : ReceiveActor
var sites = await repo.GetAllSitesAsync(ct).ConfigureAwait(false);
var contacts = new Dictionary<string, List<string>>();
// Parallel gRPC-endpoint cache fed by the SAME DB read (the streaming path's
// GrpcNodeA/BAddress columns, NOT the Akka NodeA/BAddress ones). No second poll loop —
// the gRPC transport rides this one, and the Akka transport simply ignores the field.
var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
foreach (var site in sites)
{
var addrs = new List<string>();
@@ -511,6 +575,11 @@ public class CentralCommunicationActor : ReceiveActor
}
if (addrs.Count > 0)
contacts[site.SiteIdentifier] = addrs;
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);
}
// Freeze the cross-task payload before piping to
@@ -525,82 +594,20 @@ public class CentralCommunicationActor : ReceiveActor
// address-bearing subset in `frozen`) so the aggregator prunes only
// genuinely-deleted sites and never an addressless-but-configured one.
var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
return new SiteAddressCacheLoaded(frozen, knownSiteIds);
return new SiteAddressCacheLoaded(frozen, knownSiteIds, grpcContacts);
}).PipeTo(self);
}
private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg)
{
var newSiteIds = msg.SiteContacts.Keys.ToHashSet();
var existingSiteIds = _siteClients.Keys.ToHashSet();
// Stop ClusterClients for removed sites
foreach (var removed in existingSiteIds.Except(newSiteIds))
{
_log.Info("Stopping ClusterClient for removed site {0}", removed);
Context.Stop(_siteClients[removed].Client);
_siteClients.Remove(removed);
}
// Add or update
foreach (var (siteId, addresses) in msg.SiteContacts)
{
// Parse all addresses up front inside a try/catch so a
// single malformed site row cannot abort the whole refresh loop and leave
// the cache half-updated. A bad site is logged and skipped; others proceed.
ImmutableHashSet<ActorPath> contactPaths;
try
{
contactPaths = addresses
.Select(a => ActorPath.Parse($"{a}/system/receptionist"))
.ToImmutableHashSet();
}
catch (Exception ex)
{
_log.Warning(ex,
"Malformed contact address for site {0}; skipping this site in the refresh "
+ "(other sites are unaffected)", siteId);
continue;
}
var contactStrings = addresses.ToImmutableHashSet();
// Skip if unchanged
if (_siteClients.TryGetValue(siteId, out var existing) && existing.ContactAddresses.SetEquals(contactStrings))
continue;
// Stop old client if addresses changed
if (_siteClients.ContainsKey(siteId))
{
_log.Info("Updating ClusterClient for site {0} (addresses changed)", siteId);
Context.Stop(_siteClients[siteId].Client);
// Remove now: if the replacement create below fails, a stale entry
// would route envelopes to a stopping actor.
_siteClients.Remove(siteId);
}
IActorRef client;
try
{
client = _siteClientFactory.Create(Context.System, siteId, contactPaths);
}
catch (Exception ex)
{
_log.Error(ex,
"Failed to create ClusterClient for site {0}; site is unroutable until the next refresh",
siteId);
continue;
}
_siteClients[siteId] = (client, contactStrings);
_log.Info("Created ClusterClient for site {0} with {1} contact(s)", siteId, addresses.Count);
}
_log.Info("Site ClusterClient cache refreshed with {0} site(s)", _siteClients.Count);
// Per-transport per-site resource reconciliation (create/stop ClusterClients, or
// build/drop gRPC channel pairs). Runs on the actor thread each refresh tick.
_transport.ReconcileSites(msg);
// 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.
// one refresh interval without needing a dedicated deletion event. Transport-agnostic.
_serviceProvider.GetService<ICentralHealthAggregator>()?.PruneUnknownSites(msg.KnownSiteIds);
}
@@ -673,8 +680,9 @@ public class CentralCommunicationActor : ReceiveActor
public record RefreshSiteAddresses;
/// <summary>
/// Internal message carrying the loaded site contact data from the database.
/// ClusterClient creation happens on the actor thread in HandleSiteAddressCacheLoaded.
/// 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"/>.
///
/// The payload is exposed as <see cref="IReadOnlyDictionary{TKey,TValue}"/>
/// of <see cref="IReadOnlyList{T}"/> so the Akka.NET "messages are immutable"
@@ -682,9 +690,24 @@ public record RefreshSiteAddresses;
/// discipline. The producer wraps the constructed buckets with
/// <c>List&lt;T&gt;.AsReadOnly()</c> before piping to Self.
/// </summary>
internal record SiteAddressCacheLoaded(
/// <param name="SiteContacts">Akka ClusterClient contact addresses per site (from NodeA/NodeBAddress).</param>
/// <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) — the streaming path's columns,
/// consumed by the gRPC transport and ignored by the Akka one.
/// </param>
public sealed record SiteAddressCacheLoaded(
IReadOnlyDictionary<string, IReadOnlyList<string>> SiteContacts,
IReadOnlyCollection<string> KnownSiteIds);
IReadOnlyCollection<string> KnownSiteIds,
IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts);
/// <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