diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaSiteTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaSiteTransport.cs new file mode 100644 index 00000000..8dfcdc65 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaSiteTransport.cs @@ -0,0 +1,141 @@ +using System.Collections.Immutable; +using Akka.Actor; +using Akka.Cluster.Tools.Client; +using Akka.Event; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Actors; + +/// +/// The default (and, until the Phase 1B cutover, the shipping) central→site transport: routes each +/// through a per-site Akka , exactly as +/// CentralCommunicationActor did inline before the seam was extracted. +/// +/// +/// +/// Behaviour is identical to the pre-seam code: the per-site client map is (re)built from the DB +/// refresh cache; a send to a site with no client is warned and dropped (the caller's Ask times +/// out — central never buffers); a send preserves the reply-to sender so the site's reply routes +/// straight back to the waiting Ask (or the debug-bridge actor). +/// +/// +/// Both members run on the owning actor's thread, so the client map needs no synchronisation and +/// the stored (used for Context.Stop and Context.System) +/// is only ever touched there. +/// +/// +public sealed class AkkaSiteTransport : ISiteCommandTransport +{ + private readonly ISiteClientFactory _siteClientFactory; + private readonly IActorContext _context; + private readonly ILoggingAdapter _log; + + /// + /// Per-site ClusterClient instances and their contact addresses. + /// Maps SiteIdentifier → (ClusterClient actor, set of contact address strings). + /// Refreshed by . + /// + private readonly Dictionary ContactAddresses)> _siteClients = new(); + + /// Creates the Akka transport bound to the owning actor's context. + /// Factory that creates a ClusterClient per site. + /// The owning actor's context (for Stop/System); calls stay on the actor thread. + /// The owning actor's logger, so warnings keep the actor's log source. + public AkkaSiteTransport(ISiteClientFactory siteClientFactory, IActorContext context, ILoggingAdapter log) + { + _siteClientFactory = siteClientFactory ?? throw new ArgumentNullException(nameof(siteClientFactory)); + _context = context ?? throw new ArgumentNullException(nameof(context)); + _log = log ?? throw new ArgumentNullException(nameof(log)); + } + + /// + public void Send(SiteEnvelope envelope, IActorRef replyTo) + { + ArgumentNullException.ThrowIfNull(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); + + // The Ask will timeout on the caller side — no central buffering + return; + } + + // Route via ClusterClient — replyTo is preserved for Ask response routing + entry.Client.Tell( + new ClusterClient.Send("/user/site-communication", envelope.Message), + replyTo); + } + + /// + public void ReconcileSites(SiteAddressCacheLoaded cache) + { + ArgumentNullException.ThrowIfNull(cache); + + var newSiteIds = cache.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 cache.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 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); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index 86a099a4..81adbf55 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -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; /// - /// 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 + /// ScadaBridge:Communication:SiteTransport (default ). + /// The 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. /// - private readonly Dictionary 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 /// private const string HealthReportTopic = "site-health-replica"; - /// Initializes the and wires all message handlers. + /// + /// Legacy constructor: builds the transport by reading + /// ScadaBridge:Communication:SiteTransport and wrapping + /// in an (default) or resolving the gRPC transport from + /// . Kept so the Host and the existing TestKit suites construct + /// the actor exactly as before (the factory is still the Akka seam). + /// /// DI service provider for scoped repository and aggregator access. - /// Factory used to create per-site ClusterClient actors. + /// Factory used to create per-site ClusterClient actors (Akka transport). /// /// Optional override for the audit-ingest Ask timeout; defaults to /// (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); + } + + /// + /// Primary constructor: takes the already-selected directly. + /// Used by tests that substitute the transport, and reachable via the legacy constructor. + /// + /// DI service provider for scoped repository and aggregator access. + /// The central→site command transport to route every through. + /// Optional override for the audit-ingest Ask timeout (test hook). + public CentralCommunicationActor( + IServiceProvider serviceProvider, + ISiteCommandTransport transport, + TimeSpan? auditIngestAskTimeout = null) + : this(serviceProvider, auditIngestAskTimeout) + { + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + } + + /// Shared wiring: sets scoped state and registers every message handler. The + /// is assigned by the delegating public constructor before any message + /// is dispatched. + /// DI service provider. + /// Optional audit-ingest Ask timeout override. + 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; + /// + /// Chooses the transport from ScadaBridge:Communication:SiteTransport (default + /// ). For Akka, wraps the injected + /// in an bound to this actor's + /// context; for gRPC, resolves the shared and options. + /// Runs in the constructor body, where is available. + /// + /// The DI provider carrying options and (for gRPC) the channel provider. + /// The ClusterClient factory used by the Akka transport. + /// The selected transport. + private ISiteCommandTransport SelectTransport( + IServiceProvider serviceProvider, ISiteClientFactory siteClientFactory) + { + var options = serviceProvider.GetService>()?.Value; + var kind = options?.SiteTransport ?? SiteTransportKind.Akka; + + if (kind == SiteTransportKind.Grpc) + { + var channelProvider = serviceProvider.GetRequiredService(); + var loggerFactory = serviceProvider.GetRequiredService(); + _log.Info("central→site command transport: gRPC (SiteCommandService)"); + return new Grpc.GrpcSiteTransport( + channelProvider, + options!, + loggerFactory.CreateLogger()); } - // 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>(); + // 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(); foreach (var site in sites) { var addrs = new List(); @@ -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 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()?.PruneUnknownSites(msg.KnownSiteIds); } @@ -673,8 +680,9 @@ public class CentralCommunicationActor : ReceiveActor public record RefreshSiteAddresses; /// -/// 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 . /// /// The payload is exposed as /// of so the Akka.NET "messages are immutable" @@ -682,9 +690,24 @@ public record RefreshSiteAddresses; /// discipline. The producer wraps the constructed buckets with /// List<T>.AsReadOnly() before piping to Self. /// -internal record SiteAddressCacheLoaded( +/// Akka ClusterClient contact addresses per site (from NodeA/NodeBAddress). +/// Every configured site id, address-bearing or not, for aggregator pruning. +/// +/// gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress) — the streaming path's columns, +/// consumed by the gRPC transport and ignored by the Akka one. +/// +public sealed record SiteAddressCacheLoaded( IReadOnlyDictionary> SiteContacts, - IReadOnlyCollection KnownSiteIds); + IReadOnlyCollection KnownSiteIds, + IReadOnlyDictionary GrpcContacts); + +/// +/// A site's gRPC node-pair endpoints, as loaded from Site.GrpcNodeAAddress/ +/// GrpcNodeBAddress. Either may be null when only one node has a gRPC address configured. +/// +/// NodeA gRPC base address (e.g. http://scadabridge-site-a-node-a:8083), or null. +/// NodeB gRPC base address, or null. +public readonly record struct SiteGrpcEndpoints(string? NodeA, string? NodeB); /// /// Peer-replication envelope for a site heartbeat, fanned out over the same diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs new file mode 100644 index 00000000..189c6c4e --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs @@ -0,0 +1,43 @@ +using Akka.Actor; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Actors; + +/// +/// The central→site command-send seam, injected into +/// below the handler. Exactly one implementation is active per node, +/// chosen by ScadaBridge:Communication:SiteTransport: +/// +/// — today's per-site ClusterClient path (default). +/// — the site +/// SiteCommandService gRPC plane. +/// +/// The producers above the seam (CommunicationService's 27 commands, SiteCallAuditActor's +/// 2 parked relays, and DebugStreamBridgeActor's subscribe/unsubscribe) are unchanged — they +/// still Ask/Tell a to the actor, which delegates here. +/// +/// +/// Both members run on the actor thread (from the and +/// SiteAddressCacheLoaded handlers), so implementations need no internal synchronisation for +/// their own per-site bookkeeping beyond what a background failback loop requires. +/// +public interface ISiteCommandTransport +{ + /// + /// Routes 's message to its site. Any reply the site produces is + /// delivered to — for an Ask that is the temporary ask actor + /// (completing the caller's task); for a Tell-with-sender (the debug bridge) that is the + /// originating actor. A message with no route (an unknown site) is warned and dropped so the + /// caller's Ask times out, exactly as today's ClusterClient path behaves. + /// + /// The site-addressed command envelope. + /// Where a reply (or a ) is delivered. + void Send(SiteEnvelope envelope, IActorRef replyTo); + + /// + /// Reconciles per-site transport resources (ClusterClients for Akka, channel pairs for gRPC) + /// against the freshly loaded site set. Called once per DB refresh tick with the same cache + /// message the actor already receives — the ONE DB-poll loop feeds both transports. + /// + /// The loaded site address cache (Akka contacts + gRPC endpoints + known ids). + void ReconcileSites(SiteAddressCacheLoaded cache); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index 410b3fda..0ac836e7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -1,6 +1,5 @@ namespace ZB.MOM.WW.ScadaBridge.Communication; -/// /// Which transport carries the seven site→central control messages. Selected per node by /// ScadaBridge:Communication:CentralTransport; the migration ships with /// as the default so nothing flips until a node opts in. @@ -14,6 +13,21 @@ public enum CentralTransportMode Grpc = 1, } +/// +/// Selects the transport the central→site command plane rides on. The Akka +/// per-site ClusterClient path is the default until the gRPC cutover +/// (ClusterClient→gRPC migration, Phase 1B); flipping to is the +/// rollback-by-flag switch. +/// +public enum SiteTransportKind +{ + /// Route SiteEnvelopes through the per-site Akka ClusterClient (today's default). + Akka, + + /// Route SiteEnvelopes over the site SiteCommandService gRPC plane. + Grpc +} + /// /// Configuration options for central-site communication, including per-pattern /// timeouts and transport heartbeat settings. @@ -39,6 +53,15 @@ public class CommunicationOptions /// public List CentralGrpcEndpoints { get; set; } = new(); + /// + /// Which transport the central→site command plane uses. Default + /// (the per-site ClusterClient path) — flipping to moves + /// every SiteEnvelope onto the site SiteCommandService gRPC plane. Selected inside + /// CentralCommunicationActor; CommunicationService and SiteCallAuditActor + /// are unchanged either way. Rollback at any point = flip this back to Akka. + /// + public SiteTransportKind SiteTransport { get; set; } = SiteTransportKind.Akka; + /// Timeout for deployment commands (typically longest due to apply logic). public TimeSpan DeploymentTimeout { get; set; } = TimeSpan.FromMinutes(2); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs new file mode 100644 index 00000000..18bfef06 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs @@ -0,0 +1,244 @@ +using Akka.Actor; +using Grpc.Net.Client; +using Microsoft.Extensions.Logging; +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.Actors; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// The gRPC central→site command transport: encodes each command with +/// , dials the site's SiteCommandService through the sticky +/// (PSK + x-scadabridge-site already on the channel), +/// and routes the decoded reply back to the waiting Ask (or the debug-bridge actor). +/// +/// +/// +/// Per-call deadlines match today's Ask timeouts exactly — see . +/// Behaviour is otherwise unchanged from the Akka path: an unknown/unconfigured site is warned and +/// dropped (the caller's Ask times out), and a transport fault surfaces to the caller as a +/// , which the S&F/audit layers already treat as transient. +/// +/// +/// Cross-node retry is the channel provider's job and happens only on +/// Unavailable — never on DeadlineExceeded (a write/deploy/failover may have run). +/// +/// +public sealed class GrpcSiteTransport : ISiteCommandTransport +{ + private readonly SitePairChannelProvider _channels; + private readonly CommunicationOptions _options; + private readonly ILogger _logger; + + // Sites this transport has pushed into the channel provider — used to diff removals per refresh, + // the gRPC analogue of the Akka transport's _siteClients key set. Touched only on the actor thread. + private readonly HashSet _knownSites = new(StringComparer.Ordinal); + + /// Creates the gRPC transport and ensures the provider's failback loop is running. + /// The shared per-site channel-pair provider. + /// Communication options supplying the per-command deadlines. + /// Logger for drop/fault diagnostics. + public GrpcSiteTransport( + SitePairChannelProvider channels, + CommunicationOptions options, + ILogger logger) + { + _channels = channels ?? throw new ArgumentNullException(nameof(channels)); + _options = options ?? throw new ArgumentNullException(nameof(options)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _channels.EnsureFailbackLoop(); + } + + /// + public void Send(SiteEnvelope envelope, IActorRef replyTo) + { + ArgumentNullException.ThrowIfNull(envelope); + + // UnsubscribeDebugView is the one fire-and-forget command: the site never acks it over Akka + // (the debug bridge Tells and expects nothing), so we run the RPC but do not deliver its + // synthetic ack — mirroring today's no-reply behaviour and avoiding a dead-lettered ack. + var fireAndForget = envelope.Message is UnsubscribeDebugViewRequest; + + _ = RunAsync(envelope, replyTo, fireAndForget); + } + + private async Task RunAsync(SiteEnvelope envelope, IActorRef replyTo, bool fireAndForget) + { + try + { + var reply = await SendCoreAsync(envelope, CancellationToken.None).ConfigureAwait(false); + if (!fireAndForget && !replyTo.IsNobody()) + { + replyTo.Tell(reply, ActorRefs.NoSender); + } + } + catch (SiteChannelUnavailableException) + { + // Parity with the Akka "no ClusterClient for site" path: warn and drop, so the caller's + // Ask times out. Central never buffers. + _logger.LogWarning( + "No gRPC channel for site {SiteId}; dropping {Message} (caller's Ask will time out)", + envelope.SiteId, envelope.Message.GetType().Name); + } + catch (Exception ex) + { + if (!fireAndForget && !replyTo.IsNobody()) + { + // A timeout or non-OK status faults the caller's Ask exactly as an Akka Ask timeout + // did — the S&F/audit layers already treat that as transient. + replyTo.Tell(new Status.Failure(ex), ActorRefs.NoSender); + } + else + { + _logger.LogWarning(ex, + "Fire-and-forget {Message} to site {SiteId} faulted", + envelope.Message.GetType().Name, envelope.SiteId); + } + } + } + + private Task SendCoreAsync(SiteEnvelope envelope, CancellationToken ct) + { + var message = envelope.Message; + + // Compute the absolute deadline ONCE so a cross-node failover retry shares the overall + // budget rather than restarting it. + var deadline = DateTime.UtcNow + ResolveDeadline(message); + var group = SiteCommandDtoMapper.GroupOf(message); + + return _channels.ExecuteAsync( + envelope.SiteId, + (channel, callCt) => InvokeAsync(channel, group, message, deadline, callCt), + ct); + } + + private static async Task InvokeAsync( + GrpcChannel channel, SiteCommandGroup group, object message, DateTime deadline, CancellationToken ct) + { + var client = new SiteCommandService.SiteCommandServiceClient(channel); + + switch (group) + { + case SiteCommandGroup.Lifecycle: + { + var reply = await client.ExecuteLifecycleAsync( + SiteCommandDtoMapper.ToLifecycleRequest(message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromLifecycleReply(reply); + } + + case SiteCommandGroup.OpcUa: + { + var reply = await client.ExecuteOpcUaAsync( + SiteCommandDtoMapper.ToOpcUaRequest(message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromOpcUaReply(reply); + } + + case SiteCommandGroup.Query: + { + var reply = await client.ExecuteQueryAsync( + SiteCommandDtoMapper.ToQueryRequest(message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromQueryReply(reply); + } + + case SiteCommandGroup.Parked: + { + var reply = await client.ExecuteParkedAsync( + SiteCommandDtoMapper.ToParkedRequest(message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromParkedReply(reply); + } + + case SiteCommandGroup.Route: + { + var reply = await client.ExecuteRouteAsync( + SiteCommandDtoMapper.ToRouteRequest(message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromRouteReply(reply); + } + + case SiteCommandGroup.Failover: + { + var ack = await client.TriggerFailoverAsync( + SiteCommandDtoMapper.ToProto((TriggerSiteFailover)message), + deadline: deadline, cancellationToken: ct); + return SiteCommandDtoMapper.FromProto(ack); + } + + default: + throw new ArgumentOutOfRangeException(nameof(group), group, "Unknown site command group."); + } + } + + /// + /// The per-command deadline, set EQUAL to the Ask timeout CommunicationService uses for + /// that command today, so flipping the transport changes nothing about how long a call waits. + /// Note the group is NOT a uniform deadline class: within Lifecycle, DeploymentStateQuery + /// uses QueryTimeout (not LifecycleTimeout), and TriggerSiteFailover uses + /// QueryTimeout (not LifecycleTimeout) — matching the real Ask sites, not the + /// plan's per-group table. The two parked relays (RetryParkedOperation/ + /// DiscardParkedOperation) map to QueryTimeout (30s), preserving the + /// SiteCallAuditActor inner RelayTimeout (10s) < 30s ordering. WaitForAttribute + /// keeps its dynamic request.Timeout + IntegrationTimeout budget. + /// + /// The command being sent. + /// The deadline duration for that command. + internal TimeSpan ResolveDeadline(object message) => message switch + { + RefreshDeploymentCommand => _options.DeploymentTimeout, + EnableInstanceCommand or DisableInstanceCommand or DeleteInstanceCommand => _options.LifecycleTimeout, + DeploymentStateQueryRequest => _options.QueryTimeout, + DeployArtifactsCommand => _options.ArtifactDeploymentTimeout, + + BrowseNodeCommand or SearchAddressSpaceCommand or ReadTagValuesCommand or VerifyEndpointCommand + or TrustServerCertCommand or ListServerCertsCommand or RemoveServerCertCommand or WriteTagRequest + => _options.QueryTimeout, + + EventLogQueryRequest or DebugSnapshotRequest => _options.QueryTimeout, + SubscribeDebugViewRequest or UnsubscribeDebugViewRequest => _options.DebugViewTimeout, + + ParkedMessageQueryRequest or ParkedMessageRetryRequest or ParkedMessageDiscardRequest + or RetryParkedOperation or DiscardParkedOperation + => _options.QueryTimeout, + + RouteToCallRequest or RouteToGetAttributesRequest or RouteToSetAttributesRequest + => _options.IntegrationTimeout, + RouteToWaitForAttributeRequest r => r.Timeout + _options.IntegrationTimeout, + + TriggerSiteFailover => _options.QueryTimeout, + + _ => throw new ArgumentException( + $"'{message.GetType().Name}' is not a migrated site command.", nameof(message)) + }; + + /// + public void ReconcileSites(SiteAddressCacheLoaded cache) + { + ArgumentNullException.ThrowIfNull(cache); + + var desired = cache.GrpcContacts; + + // Drop sites that lost their gRPC endpoints (or were deleted). + foreach (var removed in _knownSites.Where(s => !desired.ContainsKey(s)).ToList()) + { + _channels.RemoveSite(removed); + _knownSites.Remove(removed); + } + + // Create/refresh the rest. + foreach (var (siteId, endpoints) in desired) + { + _channels.UpdateSite(siteId, endpoints.NodeA, endpoints.NodeB); + _knownSites.Add(siteId); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs new file mode 100644 index 00000000..ee94a6a2 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -0,0 +1,436 @@ +using System.Collections.Concurrent; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// Raised when a site has no usable gRPC channel — an unknown site, or a site with neither +/// GrpcNodeAAddress nor GrpcNodeBAddress configured. The gRPC transport treats this +/// the way the Akka path treats "no ClusterClient for site": warn and drop, so the caller's Ask +/// times out (central never buffers). +/// +public sealed class SiteChannelUnavailableException(string siteId) + : Exception($"No gRPC channel is configured for site '{siteId}'.") +{ + /// The site with no configured channel. + public string SiteId { get; } = siteId; +} + +/// +/// Shared, per-site gRPC channel PAIR provider with sticky failover + failback (design §3.5). +/// Central holds one per site node (NodeA/NodeB) rather than one channel +/// with re-dial logic; NodeA is the preferred node (config order). Every call goes to the current +/// sticky channel; on connect failure / / readiness rejection +/// it flips to the other node and stays there. A background probe returns to the preferred node +/// once it answers again. Reconnect probing backs off 1 s → doubling → 60 s cap. +/// +/// +/// +/// Retry safety. retries on the other node ONLY when the first +/// attempt failed with (provably never reached a server — +/// connect refused or readiness-rejected before response headers). A +/// , or any other status, is rethrown WITHOUT a cross-node +/// retry: a WriteTag/DeployArtifacts/TriggerSiteFailover may already have +/// executed, and the layers above tolerate the one-shot ambiguity exactly as ClusterClient's Ask +/// did — the migration must not turn one write into two. +/// +/// +/// Addresses. Fed by / from the single DB +/// refresh loop (the GrpcNodeAAddress/GrpcNodeBAddress columns). PSK credentials and +/// the x-scadabridge-site header ride every channel via +/// ; removing a site invalidates its +/// cached key. +/// +/// +public sealed class SitePairChannelProvider : IDisposable +{ + private static readonly TimeSpan InitialBackoff = TimeSpan.FromSeconds(1); + private static readonly TimeSpan MaxBackoff = TimeSpan.FromSeconds(60); + private static readonly TimeSpan FailbackSweepInterval = TimeSpan.FromSeconds(1); + + private readonly ISitePskProvider _pskProvider; + private readonly CommunicationOptions _options; + private readonly ILogger _logger; + private readonly Func? _handlerFactory; + private readonly Func> _reachabilityProbe; + + private readonly ConcurrentDictionary _sites = new(StringComparer.Ordinal); + private readonly CancellationTokenSource _shutdown = new(); + private Task? _failbackLoop; + private readonly object _loopGate = new(); + private bool _disposed; + + /// Creates the provider. + /// Resolves each site's preshared key for channel credentials. + /// Communication options (keepalive settings applied to production channels). + /// Logger for failover/failback diagnostics. + /// + /// Test seam mapping an endpoint to the its channel should use + /// (e.g. an in-process TestServer handler). Null in production, where each channel builds + /// a keepalive-configured . + /// + /// + /// Test seam deciding whether a preferred node is reachable during a failback sweep. Null in + /// production, where is used. + /// + public SitePairChannelProvider( + ISitePskProvider pskProvider, + IOptions options, + ILogger logger, + Func? handlerFactory = null, + Func>? reachabilityProbe = null) + { + ArgumentNullException.ThrowIfNull(pskProvider); + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(logger); + + _pskProvider = pskProvider; + _options = options.Value; + _logger = logger; + _handlerFactory = handlerFactory; + _reachabilityProbe = reachabilityProbe ?? DefaultReachabilityProbe; + } + + /// + /// (Re)builds a site's channel pair from its gRPC node endpoints. A no-op when neither endpoint + /// changed. Disposes and rebuilds a channel whose endpoint changed, and resets stickiness to the + /// preferred node when the pair is (re)created. Called on the DB refresh tick. + /// + /// Site identifier. + /// NodeA gRPC base address, or null when unconfigured. + /// NodeB gRPC base address, or null when unconfigured. + public void UpdateSite(string siteId, string? nodeAEndpoint, string? nodeBEndpoint) + { + ArgumentException.ThrowIfNullOrWhiteSpace(siteId); + + var pair = _sites.GetOrAdd(siteId, id => new SitePair(id)); + lock (pair.Gate) + { + var changedA = !string.Equals(pair.NodeAEndpoint, nodeAEndpoint, StringComparison.Ordinal); + var changedB = !string.Equals(pair.NodeBEndpoint, nodeBEndpoint, StringComparison.Ordinal); + if (!changedA && !changedB) + { + return; + } + + if (changedA) + { + pair.ChannelA?.Dispose(); + pair.ChannelA = string.IsNullOrWhiteSpace(nodeAEndpoint) ? null : BuildChannel(nodeAEndpoint, siteId); + pair.NodeAEndpoint = nodeAEndpoint; + } + if (changedB) + { + pair.ChannelB?.Dispose(); + pair.ChannelB = string.IsNullOrWhiteSpace(nodeBEndpoint) ? null : BuildChannel(nodeBEndpoint, siteId); + pair.NodeBEndpoint = nodeBEndpoint; + } + + // Addresses changed — return to the preferred node and reset failback backoff. + pair.CurrentIsA = pair.ChannelA is not null; + pair.ResetFailback(); + _logger.LogInformation( + "gRPC channel pair for site {SiteId} refreshed (nodeA={HasA}, nodeB={HasB})", + siteId, pair.ChannelA is not null, pair.ChannelB is not null); + } + } + + /// Disposes a removed site's channels and invalidates its cached preshared key. + /// Site identifier to remove. + public void RemoveSite(string siteId) + { + if (string.IsNullOrWhiteSpace(siteId)) + { + return; + } + + if (_sites.TryRemove(siteId, out var pair)) + { + lock (pair.Gate) + { + pair.ChannelA?.Dispose(); + pair.ChannelB?.Dispose(); + pair.ChannelA = null; + pair.ChannelB = null; + } + _pskProvider.Invalidate(siteId); + _logger.LogInformation("gRPC channel pair for site {SiteId} removed", siteId); + } + } + + /// + /// Runs against the site's current sticky channel, failing over to the + /// other node once on (never on + /// ). Ensures the background failback loop is running. + /// + /// The RPC reply type. + /// Target site. + /// The RPC to run against a given channel. + /// Cancellation token. + /// The RPC reply. + /// The site has no configured channel. + public async Task ExecuteAsync( + string siteId, Func> call, CancellationToken ct) + { + ArgumentException.ThrowIfNullOrWhiteSpace(siteId); + ArgumentNullException.ThrowIfNull(call); + EnsureFailbackLoop(); + + if (!_sites.TryGetValue(siteId, out var pair)) + { + throw new SiteChannelUnavailableException(siteId); + } + + GrpcChannel primary; + GrpcChannel? secondary; + bool primaryIsA; + lock (pair.Gate) + { + primaryIsA = pair.CurrentIsA; + primary = (primaryIsA ? pair.ChannelA : pair.ChannelB) + ?? pair.ChannelA ?? pair.ChannelB + ?? throw new SiteChannelUnavailableException(siteId); + // Recompute which node `primary` actually is (the current side may be null). + primaryIsA = ReferenceEquals(primary, pair.ChannelA); + secondary = primaryIsA ? pair.ChannelB : pair.ChannelA; + } + + try + { + return await call(primary, ct).ConfigureAwait(false); + } + catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable && secondary is not null) + { + // Provably never reached a server → safe to try the other node, even for a write. + _logger.LogWarning( + "Site {SiteId} node {FromNode} unavailable; failing over to node {ToNode}", + siteId, primaryIsA ? "A" : "B", primaryIsA ? "B" : "A"); + FlipTo(pair, toIsA: !primaryIsA); + return await call(secondary, ct).ConfigureAwait(false); + } + } + + /// + /// Probes a site's preferred node (NodeA) and, if reachable, returns stickiness to it. The + /// background loop calls this per due site; tests call it to drive failback deterministically. + /// + /// Site identifier. + /// Cancellation token. + /// True when the site is now (or already) pointed at its preferred node. + internal async Task TryFailbackAsync(string siteId, CancellationToken ct) + { + if (!_sites.TryGetValue(siteId, out var pair)) + { + return false; + } + + GrpcChannel? preferred; + lock (pair.Gate) + { + if (pair.CurrentIsA || pair.ChannelA is null) + { + return true; // already on preferred (or no preferred channel to fail back to) + } + preferred = pair.ChannelA; + } + + var reachable = await _reachabilityProbe(preferred, ct).ConfigureAwait(false); + lock (pair.Gate) + { + if (reachable) + { + pair.CurrentIsA = true; + pair.ResetFailback(); + _logger.LogInformation("Site {SiteId} preferred node A reachable again; failing back", siteId); + return true; + } + + pair.BumpFailbackBackoff(); + return false; + } + } + + /// Test/diagnostic accessor: true when the site currently targets its preferred node (A). + /// Site identifier. + /// True when on NodeA (or when the site is unknown). + internal bool IsOnPreferredNode(string siteId) + => !_sites.TryGetValue(siteId, out var pair) || pair.CurrentIsA; + + /// Starts the background failback loop if not already running. Idempotent. + public void EnsureFailbackLoop() + { + if (_failbackLoop is not null || _disposed) + { + return; + } + lock (_loopGate) + { + if (_failbackLoop is null && !_disposed) + { + _failbackLoop = Task.Run(() => FailbackLoopAsync(_shutdown.Token)); + } + } + } + + private async Task FailbackLoopAsync(CancellationToken ct) + { + using var timer = new PeriodicTimer(FailbackSweepInterval); + try + { + while (await timer.WaitForNextTickAsync(ct).ConfigureAwait(false)) + { + var now = DateTime.UtcNow; + foreach (var kvp in _sites) + { + var pair = kvp.Value; + bool due; + lock (pair.Gate) + { + due = !pair.CurrentIsA && pair.ChannelA is not null && now >= pair.NextProbeUtc; + } + if (due) + { + try + { + await TryFailbackAsync(kvp.Key, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Failback probe for site {SiteId} faulted", kvp.Key); + } + } + } + } + } + catch (OperationCanceledException) + { + // Shutting down. + } + } + + private void FlipTo(SitePair pair, bool toIsA) + { + lock (pair.Gate) + { + pair.CurrentIsA = toIsA; + if (!toIsA) + { + // Failed over off the preferred node → schedule the first failback probe. + pair.ScheduleFailback(InitialBackoff); + } + else + { + pair.ResetFailback(); + } + } + } + + private GrpcChannel BuildChannel(string endpoint, string siteId) + { + var channelOptions = new GrpcChannelOptions + { + HttpHandler = _handlerFactory is not null + ? _handlerFactory(endpoint) + : new SocketsHttpHandler + { + KeepAlivePingDelay = _options.GrpcKeepAlivePingDelay, + KeepAlivePingTimeout = _options.GrpcKeepAlivePingTimeout, + KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always, + EnableMultipleHttp2Connections = true + } + }.WithSiteCredentials(_pskProvider, siteId); + + return GrpcChannel.ForAddress(endpoint, channelOptions); + } + + private static async Task DefaultReachabilityProbe(GrpcChannel channel, CancellationToken ct) + { + try + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(5)); + await channel.ConnectAsync(timeout.Token).ConfigureAwait(false); + return true; + } + catch + { + return false; + } + } + + /// + public void Dispose() + { + if (_disposed) + { + return; + } + _disposed = true; + + _shutdown.Cancel(); + try + { + _failbackLoop?.Wait(TimeSpan.FromSeconds(2)); + } + catch + { + // Best-effort loop drain on shutdown. + } + _shutdown.Dispose(); + + foreach (var pair in _sites.Values) + { + lock (pair.Gate) + { + pair.ChannelA?.Dispose(); + pair.ChannelB?.Dispose(); + } + } + _sites.Clear(); + } + + /// Per-site mutable channel state, guarded by . + private sealed class SitePair(string siteId) + { + public string SiteId { get; } = siteId; + public readonly object Gate = new(); + public string? NodeAEndpoint; + public string? NodeBEndpoint; + public GrpcChannel? ChannelA; + public GrpcChannel? ChannelB; + + /// Sticky current node; NodeA is preferred. True = pointing at NodeA. + public bool CurrentIsA = true; + + /// Next time a failback probe is due (while failed over). MaxValue = not scheduled. + public DateTime NextProbeUtc = DateTime.MaxValue; + + private TimeSpan _backoff = InitialBackoff; + + public void ScheduleFailback(TimeSpan delay) + { + _backoff = delay; + NextProbeUtc = DateTime.UtcNow + _backoff; + } + + public void BumpFailbackBackoff() + { + var next = TimeSpan.FromTicks(Math.Min(_backoff.Ticks * 2, MaxBackoff.Ticks)); + _backoff = next; + NextProbeUtc = DateTime.UtcNow + _backoff; + } + + public void ResetFailback() + { + _backoff = InitialBackoff; + NextProbeUtc = DateTime.MaxValue; + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj b/src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj index 03c45753..ccc4bf18 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ZB.MOM.WW.ScadaBridge.Communication.csproj @@ -10,6 +10,9 @@ + + diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index 96a495b4..433d2990 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -167,6 +167,14 @@ try }); builder.Services.AddSingleton< ZB.MOM.WW.ScadaBridge.Communication.Grpc.CentralControlGrpcService>(); + + // Shared per-site gRPC channel-pair provider (sticky failover/failback) backing the + // GrpcSiteTransport when ScadaBridge:Communication:SiteTransport=Grpc. Central-only, and a + // no-op until CentralCommunicationActor builds the gRPC transport (its failback loop starts + // on first use), so registering it unconditionally is harmless under the default Akka path. + builder.Services.AddSingleton< + ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitePairChannelProvider>(); + builder.Services.AddHealthMonitoring(); builder.Services.AddCentralHealthAggregation(); builder.Services.AddExternalSystemGateway(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs index cb208fb5..c202c248 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs @@ -39,7 +39,8 @@ public class CentralCommunicationActorClientLifecycleTests : TestKit private static SiteAddressCacheLoaded Load(string siteId, params string[] addrs) => new(new Dictionary> { [siteId] = addrs.ToList().AsReadOnly() }, - new[] { siteId }); + new[] { siteId }, + new Dictionary()); [Fact] public void PeriodicRefresh_PrunesDeletedSites_FromHealthAggregator() diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs new file mode 100644 index 00000000..a7a4770f --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs @@ -0,0 +1,136 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// T1B.3 seam tests: routing a +/// through an injected (transport-agnostic), proving the +/// envelope reaches the transport, the reply routes back to the waiting Ask, and the DB refresh +/// tick reconciles per-site transport resources. +/// +public class CentralCommunicationActorTransportTests : TestKit +{ + public CentralCommunicationActorTransportTests() : base(@"akka.loglevel = WARNING") { } + + private static Site GrpcSite(string id, string? grpcA = "http://a:8083", string? grpcB = "http://b:8083") => + new("Test " + id, id) + { + NodeAAddress = $"akka.tcp://scadabridge@{id}-a:8081", + GrpcNodeAAddress = grpcA, + GrpcNodeBAddress = grpcB + }; + + private (IActorRef actor, ISiteCommandTransport transport, ISiteRepository repo) CreateActor( + IEnumerable? sites = null) + { + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(sites?.ToList() ?? new List()); + + var services = new ServiceCollection(); + services.AddScoped(_ => repo); + var sp = services.BuildServiceProvider(); + + var transport = Substitute.For(); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + return (actor, transport, repo); + } + + [Fact] + public void SiteEnvelope_IsRoutedToTheTransport_WithTheSenderAsReplyTo() + { + var (actor, transport, _) = CreateActor(); + var probe = CreateTestProbe(); + + var cmd = new EnableInstanceCommand("cmd-1", "Site1.Pump1", DateTimeOffset.UtcNow); + actor.Tell(new SiteEnvelope("site-a", cmd), probe.Ref); + + AwaitAssert(() => transport.Received(1).Send( + Arg.Is(e => e.SiteId == "site-a" && ReferenceEquals(e.Message, cmd)), + probe.Ref)); + } + + [Fact] + public async Task AskReply_FromTransport_RoutesBackToTheWaitingAsk() + { + var (actor, transport, _) = CreateActor(); + + // The transport substitute stands in for the site: when handed the envelope it delivers a + // reply to the captured replyTo, exactly as a real transport pipes the site's reply back. + var reply = new DeploymentStatusResponse( + "dep-1", "Site1.Pump1", DeploymentStatus.Success, null, DateTimeOffset.UtcNow); + transport + .When(t => t.Send(Arg.Any(), Arg.Any())) + .Do(ci => ci.Arg().Tell(reply)); + + var cmd = new RefreshDeploymentCommand( + "dep-1", "Site1.Pump1", "hash", "multi-role", DateTimeOffset.UtcNow, "http://c:5000", "tok"); + var result = await actor.Ask( + new SiteEnvelope("site-a", cmd), TimeSpan.FromSeconds(3)); + + Assert.Equal("dep-1", result.DeploymentId); + Assert.Equal(DeploymentStatus.Success, result.Status); + } + + [Fact] + public void DbRefresh_ReconcilesTheTransport_WithTheLoadedGrpcEndpoints() + { + var (_, transport, _) = CreateActor(new[] { GrpcSite("site-a") }); + + // PreStart fires the refresh at Zero; the loaded cache is handed to the transport. + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(c => + c.KnownSiteIds.Contains("site-a") + && c.GrpcContacts.ContainsKey("site-a") + && c.GrpcContacts["site-a"].NodeA == "http://a:8083" + && c.GrpcContacts["site-a"].NodeB == "http://b:8083")), + TimeSpan.FromSeconds(3)); + } + + [Fact] + public void SiteAddedAndRemoved_AcrossRefreshes_FlowsThroughToTheTransport() + { + var (actor, transport, repo) = CreateActor(new[] { GrpcSite("site-a") }); + + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(c => c.GrpcContacts.ContainsKey("site-a"))), + TimeSpan.FromSeconds(3)); + + // site-a removed, site-b added. + repo.GetAllSitesAsync(Arg.Any()).Returns(new List { GrpcSite("site-b") }); + actor.Tell(new RefreshSiteAddresses()); + + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(c => + c.GrpcContacts.ContainsKey("site-b") + && !c.GrpcContacts.ContainsKey("site-a") + && c.KnownSiteIds.Contains("site-b") + && !c.KnownSiteIds.Contains("site-a"))), + TimeSpan.FromSeconds(3)); + } + + [Fact] + public void SiteWithoutGrpcAddresses_IsAbsentFromGrpcContacts_ButStillKnown() + { + // A site with only Akka addresses (no gRPC columns) is a known site but carries no gRPC + // endpoint — the gRPC transport must not try to dial it, the Akka one still can. + var akkaOnly = new Site("Akka only", "site-x") + { + NodeAAddress = "akka.tcp://scadabridge@site-x-a:8081" + }; + var (_, transport, _) = CreateActor(new[] { akkaOnly }); + + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(c => + c.KnownSiteIds.Contains("site-x") && !c.GrpcContacts.ContainsKey("site-x"))), + TimeSpan.FromSeconds(3)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportDeadlineTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportDeadlineTests.cs new file mode 100644 index 00000000..44f79d1d --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportDeadlineTests.cs @@ -0,0 +1,107 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +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.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// Pins to the EXACT Ask timeout +/// uses for each command today, so flipping the transport cannot +/// change how long any call waits. Every timeout is given a distinct value so a wrong mapping is +/// caught, not masked by two defaults that happen to be equal (QueryTimeout == IntegrationTimeout in +/// production). +/// +public class GrpcSiteTransportDeadlineTests +{ + private static readonly CommunicationOptions Opts = new() + { + DeploymentTimeout = TimeSpan.FromSeconds(120), + LifecycleTimeout = TimeSpan.FromSeconds(30), + ArtifactDeploymentTimeout = TimeSpan.FromSeconds(60), + QueryTimeout = TimeSpan.FromSeconds(31), + IntegrationTimeout = TimeSpan.FromSeconds(32), + DebugViewTimeout = TimeSpan.FromSeconds(10) + }; + + private static readonly GrpcSiteTransport Transport = new( + new SitePairChannelProvider( + new NoKeyProvider(), Options.Create(Opts), NullLogger.Instance), + Opts, + NullLogger.Instance); + + private static readonly DateTimeOffset T = DateTimeOffset.UtcNow; + + public static IEnumerable Cases() + { + // Lifecycle group — note it is NOT one deadline class. + yield return [new RefreshDeploymentCommand("d", "i", "h", "by", T, "u", "t"), Opts.DeploymentTimeout]; + yield return [new EnableInstanceCommand("c", "i", T), Opts.LifecycleTimeout]; + yield return [new DisableInstanceCommand("c", "i", T), Opts.LifecycleTimeout]; + yield return [new DeleteInstanceCommand("c", "i", T), Opts.LifecycleTimeout]; + // DeploymentStateQuery uses QueryTimeout in CommunicationService, NOT LifecycleTimeout. + yield return [new DeploymentStateQueryRequest("c", "i", T), Opts.QueryTimeout]; + yield return [new DeployArtifactsCommand("d", null, null, null, null, null, null, T), Opts.ArtifactDeploymentTimeout]; + + // OPC UA group — all QueryTimeout. + yield return [new BrowseNodeCommand("conn", null, null, null), Opts.QueryTimeout]; + yield return [new SearchAddressSpaceCommand("conn", "q", 1, 1, null), Opts.QueryTimeout]; + yield return [new ReadTagValuesCommand("conn", []), Opts.QueryTimeout]; + yield return [new VerifyEndpointCommand("conn", "OpcUa", "{}", null), Opts.QueryTimeout]; + yield return [new TrustServerCertCommand("conn", "ZGVy", "AA", null), Opts.QueryTimeout]; + yield return [new ListServerCertsCommand(null), Opts.QueryTimeout]; + yield return [new RemoveServerCertCommand("AA", null), Opts.QueryTimeout]; + yield return [new WriteTagRequest("c", "conn", "tag", 1, T), Opts.QueryTimeout]; + + // Query group. + yield return [new EventLogQueryRequest("c", "s", null, null, null, null, null, null, null, 10, T), Opts.QueryTimeout]; + yield return [new DebugSnapshotRequest("i", "c"), Opts.QueryTimeout]; + yield return [new SubscribeDebugViewRequest("i", "c"), Opts.DebugViewTimeout]; + yield return [new UnsubscribeDebugViewRequest("i", "c"), Opts.DebugViewTimeout]; + + // Parked group — the two relays keep QueryTimeout (30s) so SiteCallAudit's inner + // RelayTimeout (10s) still expires first. + yield return [new ParkedMessageQueryRequest("c", "s", 1, 10, T), Opts.QueryTimeout]; + yield return [new ParkedMessageRetryRequest("c", "s", "m", T), Opts.QueryTimeout]; + yield return [new ParkedMessageDiscardRequest("c", "s", "m", T), Opts.QueryTimeout]; + yield return [new RetryParkedOperation("c", new TrackedOperationId(Guid.NewGuid())), Opts.QueryTimeout]; + yield return [new DiscardParkedOperation("c", new TrackedOperationId(Guid.NewGuid())), Opts.QueryTimeout]; + + // Route group. + yield return [new RouteToCallRequest("c", "i", "m", null, T), Opts.IntegrationTimeout]; + yield return [new RouteToGetAttributesRequest("c", "i", [], T), Opts.IntegrationTimeout]; + yield return [new RouteToSetAttributesRequest("c", "i", new Dictionary(), T), Opts.IntegrationTimeout]; + + // Failover — QueryTimeout in CommunicationService, NOT LifecycleTimeout. + yield return [new TriggerSiteFailover("c", "s"), Opts.QueryTimeout]; + } + + [Theory] + [MemberData(nameof(Cases))] + public void ResolveDeadline_MatchesTodaysAskTimeout(object command, TimeSpan expected) + { + Assert.Equal(expected, Transport.ResolveDeadline(command)); + } + + [Fact] + public void WaitForAttribute_UsesItsDynamicTimeoutPlusIntegrationSlack() + { + // CommunicationService.RouteToWaitForAttributeAsync uses request.Timeout + IntegrationTimeout. + var wait = new RouteToWaitForAttributeRequest("c", "i", "attr", "10", TimeSpan.FromSeconds(45), T); + Assert.Equal(TimeSpan.FromSeconds(45) + Opts.IntegrationTimeout, Transport.ResolveDeadline(wait)); + } + + private sealed class NoKeyProvider : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new("k"); + public void Invalidate(string siteId) { } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs new file mode 100644 index 00000000..ccc0df1e --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs @@ -0,0 +1,218 @@ +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// T1B.3: failover/failback + credential/deadline proof over +/// two in-process gRPC s (NodeA preferred, NodeB standby). Exercises the +/// real client stack — PSK call credentials, per-call deadline, sticky failover, no-retry on +/// , and failback to the preferred node. +/// +public sealed class GrpcSiteTransportFailoverTests : IAsyncLifetime +{ + private const string SiteId = "site-1"; + private const string SiteKey = "the-site-1-key"; + private const string EndpointA = "http://node-a"; + private const string EndpointB = "http://node-b"; + + private IHost _hostA = null!; + private IHost _hostB = null!; + private StubSiteCommandService _stubA = null!; + private StubSiteCommandService _stubB = null!; + private SitePairChannelProvider _provider = null!; + private volatile bool _preferredReachable; + + /// + public async Task InitializeAsync() + { + (_hostA, _stubA) = await StartServerAsync(); + (_hostB, _stubB) = await StartServerAsync(); + + var handlerA = _hostA.GetTestServer().CreateHandler(); + var handlerB = _hostB.GetTestServer().CreateHandler(); + + _provider = new SitePairChannelProvider( + new FixedPskProvider(SiteKey), + Options.Create(new CommunicationOptions()), + NullLogger.Instance, + handlerFactory: endpoint => endpoint == EndpointA ? handlerA : handlerB, + reachabilityProbe: (_, _) => Task.FromResult(_preferredReachable)); + + _provider.UpdateSite(SiteId, EndpointA, EndpointB); + } + + /// + public async Task DisposeAsync() + { + _provider.Dispose(); + await _hostA.StopAsync(); + await _hostB.StopAsync(); + _hostA.Dispose(); + _hostB.Dispose(); + } + + private static async Task<(IHost, StubSiteCommandService)> StartServerAsync() + { + var stub = new StubSiteCommandService(); + var host = await new HostBuilder() + .ConfigureWebHost(web => web + .UseTestServer() + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(stub); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapGrpcService()); + })) + .StartAsync(); + return (host, stub); + } + + private static LifecycleRequest EnableReq() => + SiteCommandDtoMapper.ToLifecycleRequest( + new EnableInstanceCommand("cmd-1", "Site1.Pump1", DateTimeOffset.UtcNow)); + + private Task CallAsync(TimeSpan? deadline = null) => + _provider.ExecuteAsync( + SiteId, + (channel, ct) => + { + var client = new SiteCommandService.SiteCommandServiceClient(channel); + return client.ExecuteLifecycleAsync( + EnableReq(), + deadline: DateTime.UtcNow + (deadline ?? TimeSpan.FromSeconds(10)), + cancellationToken: ct).ResponseAsync; + }, + CancellationToken.None); + + [Fact] + public async Task HealthyCall_HitsPreferredNodeA_WithPskAndSiteHeaderAndDeadline() + { + var reply = await CallAsync(); + + Assert.Equal(LifecycleReply.ReplyOneofCase.InstanceLifecycle, reply.ReplyCase); + Assert.Equal(1, _stubA.LifecycleCalls); + Assert.Equal(0, _stubB.LifecycleCalls); + Assert.Equal($"Bearer {SiteKey}", _stubA.LastAuthHeader); + Assert.Equal(SiteId, _stubA.LastSiteHeader); + Assert.True(_stubA.DeadlineWasSet, "the client must set a per-call deadline"); + Assert.True(_provider.IsOnPreferredNode(SiteId)); + } + + [Fact] + public async Task Unavailable_OnNodeA_FailsOverToNodeB_AndStaysSticky() + { + _stubA.ThrowUnavailable = true; + + var reply = await CallAsync(); + + // Failed over: A was tried (and threw), B answered. + Assert.Equal(LifecycleReply.ReplyOneofCase.InstanceLifecycle, reply.ReplyCase); + Assert.Equal(1, _stubA.LifecycleCalls); + Assert.Equal(1, _stubB.LifecycleCalls); + Assert.False(_provider.IsOnPreferredNode(SiteId)); + + // Sticky: the next call goes straight to B without re-touching A. + await CallAsync(); + Assert.Equal(1, _stubA.LifecycleCalls); + Assert.Equal(2, _stubB.LifecycleCalls); + } + + [Fact] + public async Task Failback_ReturnsToPreferredNodeA_OncePreferredIsReachableAgain() + { + _stubA.ThrowUnavailable = true; + await CallAsync(); // flips to B + Assert.False(_provider.IsOnPreferredNode(SiteId)); + + // NodeA recovers; the failback probe reports it reachable. + _stubA.ThrowUnavailable = false; + _preferredReachable = true; + var back = await _provider.TryFailbackAsync(SiteId, CancellationToken.None); + + Assert.True(back); + Assert.True(_provider.IsOnPreferredNode(SiteId)); + + var beforeA = _stubA.LifecycleCalls; + await CallAsync(); + Assert.Equal(beforeA + 1, _stubA.LifecycleCalls); // next call back on A + } + + [Fact] + public async Task DeadlineExceeded_OnNodeA_IsNotRetriedOnNodeB() + { + // A DeadlineExceeded is ambiguous (a WriteTag/Deploy/Failover may already have executed), so + // — unlike Unavailable — it must NOT fail over to B. Modelled by A returning the status + // directly, isolating the retry-decision from TestServer's own timeout mechanics. + _stubA.StatusToThrow = StatusCode.DeadlineExceeded; + + var ex = await Assert.ThrowsAsync(() => CallAsync()); + + Assert.Equal(StatusCode.DeadlineExceeded, ex.StatusCode); + Assert.Equal(1, _stubA.LifecycleCalls); + Assert.Equal(0, _stubB.LifecycleCalls); // B was never tried + Assert.True(_provider.IsOnPreferredNode(SiteId), "a deadline must not flip stickiness"); + } + + [Fact] + public async Task UnknownSite_Throws_SiteChannelUnavailable() + { + await Assert.ThrowsAsync( + () => _provider.ExecuteAsync( + "not-configured", (_, _) => Task.FromResult(new LifecycleReply()), CancellationToken.None)); + } + + private sealed class FixedPskProvider(string key) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new(key); + public void Invalidate(string siteId) { } + } + + /// Stub SiteCommandService recording what the client sent and letting a test steer faults. + private sealed class StubSiteCommandService : SiteCommandService.SiteCommandServiceBase + { + private int _lifecycleCalls; + public int LifecycleCalls => Volatile.Read(ref _lifecycleCalls); + public string? LastAuthHeader { get; private set; } + public string? LastSiteHeader { get; private set; } + public bool DeadlineWasSet { get; private set; } + public volatile bool ThrowUnavailable; + public StatusCode? StatusToThrow; + + public override Task ExecuteLifecycle(LifecycleRequest request, ServerCallContext context) + { + Interlocked.Increment(ref _lifecycleCalls); + LastAuthHeader = context.RequestHeaders + .FirstOrDefault(h => h.Key == ControlPlaneCredentials.AuthorizationHeader)?.Value; + LastSiteHeader = context.RequestHeaders + .FirstOrDefault(h => h.Key == ControlPlaneCredentials.SiteHeader)?.Value; + DeadlineWasSet = context.Deadline != DateTime.MaxValue; + + if (ThrowUnavailable) + { + throw new RpcException(new Status(StatusCode.Unavailable, "node down")); + } + if (StatusToThrow is { } status) + { + throw new RpcException(new Status(status, "modelled fault")); + } + + return Task.FromResult(SiteCommandDtoMapper.ToLifecycleReply( + new InstanceLifecycleResponse("cmd-1", "Site1.Pump1", true, null, DateTimeOffset.UtcNow))); + } + } +}