diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaCentralTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaCentralTransport.cs new file mode 100644 index 00000000..a2d9befc --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaCentralTransport.cs @@ -0,0 +1,185 @@ +using Akka.Actor; +using Akka.Cluster.Tools.Client; +using Akka.Event; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Actors; + +/// +/// The that carries the seven site→central sends over Akka +/// ClusterClient — the transport in production today, and the default. Every method is a +/// verbatim lift of the corresponding SiteCommunicationActor send block: it forwards a +/// to /user/central-communication with the +/// as the send's sender, so central's reply routes straight back to the +/// waiting Ask rather than through the site communication actor. +/// +/// +/// The ClusterClient reference arrives after construction via +/// (the actor forwards the RegisterCentralClient message it +/// receives once the Host builds the client). Until then — and if central contact points are not +/// configured at all — the client is null and each method answers the same transient-failure reply +/// the old inline handlers did. +/// +public sealed class AkkaCentralTransport : ICentralTransport +{ + /// The receptionist-registered path of the central communication actor. + private const string CentralPath = "/user/central-communication"; + + private readonly ILoggingAdapter? _log; + private IActorRef? _centralClient; + + /// Creates the transport with no logging adapter (behaviourally identical; warnings are dropped). + public AkkaCentralTransport() + { + } + + /// Creates the transport bound to the site communication actor's logging adapter. + /// Logging adapter used for the "no ClusterClient registered" warnings. + public AkkaCentralTransport(ILoggingAdapter log) + { + _log = log; + } + + /// + /// Registers the central ClusterClient once the Host has built it. Called from the site + /// communication actor's RegisterCentralClient handler. + /// + /// The ClusterClient reaching the central cluster. + public void SetCentralClient(IActorRef centralClient) + { + _centralClient = centralClient; + _log?.Info("Registered central ClusterClient"); + } + + /// + public void SubmitNotification(NotificationSubmit message, IActorRef replyTo) + { + if (_centralClient == null) + { + // No ClusterClient registered yet (e.g. central contact points not + // configured, or registration not yet completed). A non-accepted ack + // makes the S&F forwarder treat this as transient and retry later. + _log?.Warning( + "Cannot forward NotificationSubmit {0} — no central ClusterClient registered", + message.NotificationId); + replyTo.Tell(new NotificationSubmitAck( + message.NotificationId, Accepted: false, Error: "Central ClusterClient not registered")); + return; + } + + _log?.Debug("Forwarding NotificationSubmit {0} to central", message.NotificationId); + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo) + { + if (_centralClient == null) + { + // No ClusterClient registered yet. Reply Found: false so Notify.Status + // falls back to the site S&F buffer to decide Forwarding vs Unknown. + _log?.Warning( + "Cannot forward NotificationStatusQuery {0} — no central ClusterClient registered", + message.NotificationId); + replyTo.Tell(new NotificationStatusResponse( + message.CorrelationId, Found: false, Status: "Unknown", + RetryCount: 0, LastError: null, DeliveredAt: null)); + return; + } + + _log?.Debug("Forwarding NotificationStatusQuery {0} to central", message.NotificationId); + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo) + { + if (_centralClient == null) + { + // No ClusterClient registered yet. Faulting the Ask makes the + // SiteAuditTelemetryActor drain loop treat this as transient and keep + // the rows Pending for the next tick. + _log?.Warning( + "Cannot forward IngestAuditEventsCommand ({0} events) — no central ClusterClient registered", + message.Events.Count); + replyTo.Tell(new Status.Failure( + new InvalidOperationException("Central ClusterClient not registered"))); + return; + } + + _log?.Debug("Forwarding IngestAuditEventsCommand ({0} events) to central", message.Events.Count); + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo) + { + if (_centralClient == null) + { + _log?.Warning( + "Cannot forward IngestCachedTelemetryCommand ({0} entries) — no central ClusterClient registered", + message.Entries.Count); + replyTo.Tell(new Status.Failure( + new InvalidOperationException("Central ClusterClient not registered"))); + return; + } + + _log?.Debug("Forwarding IngestCachedTelemetryCommand ({0} entries) to central", message.Entries.Count); + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo) + { + if (_centralClient == null) + { + // No ClusterClient registered yet. Faulting the Ask makes the + // SiteReconciliationActor treat the pass as best-effort-failed; it + // logs a warning and retries reconcile on the next node startup. + _log?.Warning( + "Cannot forward ReconcileSiteRequest for site {0} node {1} — no central ClusterClient registered", + message.SiteIdentifier, message.NodeId); + replyTo.Tell(new Status.Failure( + new InvalidOperationException("Central ClusterClient not registered"))); + return; + } + + _log?.Debug( + "Forwarding ReconcileSiteRequest for site {0} node {1} ({2} local instance(s)) to central", + message.SiteIdentifier, message.NodeId, message.LocalNameToRevisionHash.Count); + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo) + { + if (_centralClient == null) + { + // No ClusterClient registered yet. A non-accepted ack makes the + // sender's counter-restore path treat this tick as a loss. + _log?.Warning( + "Cannot forward SiteHealthReport #{0} — no central ClusterClient registered", + message.SequenceNumber); + replyTo.Tell(new SiteHealthReportAck( + message.SiteId, message.SequenceNumber, Accepted: false, + Error: "Central ClusterClient not registered")); + return; + } + + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), replyTo); + } + + /// + public void SendHeartbeat(HeartbeatMessage message, IActorRef self) + { + if (_centralClient == null) + { + return; + } + + _centralClient.Tell(new ClusterClient.Send(CentralPath, message), self); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ICentralTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ICentralTransport.cs new file mode 100644 index 00000000..99ef0977 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ICentralTransport.cs @@ -0,0 +1,78 @@ +using Akka.Actor; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Actors; + +/// +/// The site→central transport seam: one method per the seven messages +/// sends to /user/central-communication today. +/// +/// +/// +/// The actor's receive handlers no longer own the wire plumbing — they capture the current +/// Sender and hand it to the transport as . Two implementations +/// exist behind the ScadaBridge:Communication:CentralTransport flag: the default +/// (verbatim of the old ClusterClient.Send path, +/// including the exact sender-forwarding that routes central's reply straight back to the waiting +/// Ask) and (a gRPC dial of CentralControlService). +/// +/// +/// Reply/fault contract, identical on both transports. Each Ask-returning method (all but +/// the heartbeat) guarantees exactly one reply eventually lands at : +/// either the real reply type the caller Asks for, or a transient-failure signal. The Akka path +/// sends a not-accepted ack / when no ClusterClient is registered; +/// the gRPC path sends on any non-OK status (a timeout or an +/// Unavailable that could not be failed over). Both are what the S&F / audit / health +/// layers above the seam already treat as transient — rows stay buffered, counters restore, the +/// pass re-runs. +/// +/// +/// The heartbeat stays fire-and-forget end-to-end. takes no +/// replyTo and never surfaces a fault: a transport failure is swallowed and logged, exactly +/// as the old Tell dropped it. A failing heartbeat must never fault the site's heartbeat +/// timer path. +/// +/// +public interface ICentralTransport +{ + /// Forwards a buffered notification; central replies to . + /// The notification submission. + /// The actor (the S&F forwarder's Ask) the ack routes back to. + void SubmitNotification(NotificationSubmit message, IActorRef replyTo); + + /// Forwards a Notify.Status query; central replies to . + /// The status query. + /// The actor (the Notify helper's Ask) the response routes back to. + void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo); + + /// Pushes a batch of audit events; central replies to . + /// The audit-event ingest command. + /// The actor (the telemetry drain's Ask) the reply routes back to. + void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo); + + /// Pushes a batch of combined cached-call telemetry; central replies to . + /// The cached-telemetry ingest command. + /// The actor (the telemetry drain's Ask) the reply routes back to. + void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo); + + /// Reports a node's startup inventory; central replies to . + /// The reconcile request. + /// The actor (the reconciliation Ask) the response routes back to. + void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo); + + /// Reports periodic site health; central replies to . + /// The health report. + /// The actor (the health transport's Ask) the ack routes back to. + void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo); + + /// + /// Sends an application heartbeat, fire-and-forget. Never replies and never faults; a failure + /// is swallowed and logged. + /// + /// The heartbeat. + /// The site communication actor, used as the sender on the Akka path (ignored on gRPC). + void SendHeartbeat(HeartbeatMessage message, IActorRef self); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs index 8199b80d..b8a2a6f0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs @@ -1,6 +1,5 @@ using Akka.Actor; using Akka.Cluster; -using Akka.Cluster.Tools.Client; using Akka.Event; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; @@ -45,10 +44,14 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers private readonly IActorRef _deploymentManagerProxy; /// - /// ClusterClient reference for sending messages to the central cluster. - /// Set via RegisterCentralClient message. + /// The site→central transport. Finalized in to the injected instance, + /// or a default (ClusterClient) when none is supplied — so + /// the seven site→central sends delegate here rather than owning the wire plumbing inline. /// - private IActorRef? _centralClient; + private ICentralTransport _transport; + + /// The transport supplied by the Host (null selects the default Akka transport). + private readonly ICentralTransport? _injectedTransport; /// /// Local actor references for routing specific message patterns. @@ -73,24 +76,36 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers /// pass a stub so they do not need to load Akka.Cluster into the TestKit /// ActorSystem. /// + /// + /// The site→central transport. null (the default, used by every existing test and by + /// the Host's Akka path) selects an over ClusterClient; the + /// Host injects a when + /// ScadaBridge:Communication:CentralTransport is Grpc. + /// public SiteCommunicationActor( string siteId, CommunicationOptions options, IActorRef deploymentManagerProxy, Func? isActiveCheck = null, - Func? failOverRole = null) + Func? failOverRole = null, + ICentralTransport? transport = null) { _siteId = siteId; _options = options; _deploymentManagerProxy = deploymentManagerProxy; _isActiveCheck = isActiveCheck ?? DefaultIsActiveCheck; _failOverRole = failOverRole ?? DefaultFailOverRole; + _injectedTransport = transport; + // Finalized in PreStart (where _log is usable for the default transport); assigned here + // too so the field is definitely-assigned for the constructor's Receive closures. + _transport = transport!; - // Registration + // Registration. Feeding the ClusterClient into the transport is a no-op unless the + // default/Akka transport is in use — the gRPC transport dials configured endpoints and + // never receives this message (the Host does not create a ClusterClient for it). Receive(msg => { - _centralClient = msg.Client; - _log.Info("Registered central ClusterClient"); + (_transport as AkkaCentralTransport)?.SetCentralClient(msg.Client); }); Receive(HandleRegisterLocalHandler); @@ -274,157 +289,33 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers // from cluster state, not from who received the message. Receive(HandleTriggerSiteFailover); - // Notification Outbox: forward a buffered notification submitted by the site - // Store-and-Forward Engine to the central cluster. The original Sender (the - // S&F forwarder's Ask) is forwarded as the ClusterClient.Send sender so the - // NotificationSubmitAck routes straight back to the waiting Ask, not here. - Receive(msg => - { - if (_centralClient == null) - { - // No ClusterClient registered yet (e.g. central contact points not - // configured, or registration not yet completed). A non-accepted ack - // makes the S&F forwarder treat this as transient and retry later. - _log.Warning( - "Cannot forward NotificationSubmit {0} — no central ClusterClient registered", - msg.NotificationId); - Sender.Tell(new NotificationSubmitAck( - msg.NotificationId, Accepted: false, Error: "Central ClusterClient not registered")); - return; - } + // The seven site→central sends now delegate to the injected transport (ClusterClient by + // default, gRPC when configured). Each handler captures the current Sender as the reply + // target so central's reply routes straight back to the waiting Ask, not through this + // actor — the exact sender-forwarding the ClusterClient path relied on. The per-message + // "no transport / not-accepted" fallbacks live inside the transport now. - _log.Debug("Forwarding NotificationSubmit {0} to central", msg.NotificationId); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); + // Notification Outbox: forward a buffered notification (S&F forwarder's Ask → ack back). + Receive(msg => _transport.SubmitNotification(msg, Sender)); - // Notification Outbox: forward a Notify.Status query to the central cluster. - // The original Sender (the Notify helper's Ask) is forwarded as the - // ClusterClient.Send sender so the NotificationStatusResponse routes straight - // back to the waiting Ask, not here. - Receive(msg => - { - if (_centralClient == null) - { - // No ClusterClient registered yet. Reply Found: false so Notify.Status - // falls back to the site S&F buffer to decide Forwarding vs Unknown. - _log.Warning( - "Cannot forward NotificationStatusQuery {0} — no central ClusterClient registered", - msg.NotificationId); - Sender.Tell(new NotificationStatusResponse( - msg.CorrelationId, Found: false, Status: "Unknown", - RetryCount: 0, LastError: null, DeliveredAt: null)); - return; - } + // Notification Outbox: forward a Notify.Status query (Notify helper's Ask → response back). + Receive(msg => _transport.QueryNotificationStatus(msg, Sender)); - _log.Debug("Forwarding NotificationStatusQuery {0} to central", msg.NotificationId); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); + // Audit Log: forward a batch of site-local audit events (telemetry drain's Ask → reply back). + Receive(msg => _transport.IngestAuditEvents(msg, Sender)); - // Audit Log: forward a batch of site-local audit events to the - // central cluster. The site SiteAuditTelemetryActor drains its SQLite - // Pending queue through the ClusterClientSiteAuditClient, which Asks - // this actor; the original Sender (that Ask) is passed as the - // ClusterClient.Send sender so the IngestAuditEventsReply routes - // straight back to the waiting Ask, not here. Mirrors NotificationSubmit. - Receive(msg => - { - if (_centralClient == null) - { - // No ClusterClient registered yet (e.g. central contact points - // not configured, or registration not yet completed). Faulting - // the Ask makes the SiteAuditTelemetryActor drain loop treat - // this as transient and keep the rows Pending for the next tick. - _log.Warning( - "Cannot forward IngestAuditEventsCommand ({0} events) — no central ClusterClient registered", - msg.Events.Count); - Sender.Tell(new Status.Failure( - new InvalidOperationException("Central ClusterClient not registered"))); - return; - } + // Audit Log: forward a batch of combined cached-call telemetry (telemetry drain's Ask → reply back). + Receive(msg => _transport.IngestCachedTelemetry(msg, Sender)); - _log.Debug("Forwarding IngestAuditEventsCommand ({0} events) to central", msg.Events.Count); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); + // Site startup reconciliation: forward the node's local-inventory request (reconcile Ask → response back). + Receive(msg => _transport.ReconcileSite(msg, Sender)); - // Audit Log: forward a batch of combined cached-call telemetry - // packets to the central cluster. Same forward + reply-routing pattern - // as IngestAuditEventsCommand; central replies with an - // IngestCachedTelemetryReply. - Receive(msg => - { - if (_centralClient == null) - { - _log.Warning( - "Cannot forward IngestCachedTelemetryCommand ({0} entries) — no central ClusterClient registered", - msg.Entries.Count); - Sender.Tell(new Status.Failure( - new InvalidOperationException("Central ClusterClient not registered"))); - return; - } - - _log.Debug("Forwarding IngestCachedTelemetryCommand ({0} entries) to central", msg.Entries.Count); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); - - // Site startup reconciliation: forward the node's local-inventory - // ReconcileSiteRequest to the central cluster. The original Sender (the - // SiteReconciliationActor's Ask) is passed as the ClusterClient.Send sender so - // the ReconcileSiteResponse routes straight back to the waiting Ask, not here. - // Mirrors IngestAuditEventsCommand. - Receive(msg => - { - if (_centralClient == null) - { - // No ClusterClient registered yet (e.g. central contact points not - // configured, or registration not yet completed). Faulting the Ask makes - // the SiteReconciliationActor treat the pass as best-effort-failed; it - // logs a warning and retries reconcile on the next node startup. - _log.Warning( - "Cannot forward ReconcileSiteRequest for site {0} node {1} — no central ClusterClient registered", - msg.SiteIdentifier, msg.NodeId); - Sender.Tell(new Status.Failure( - new InvalidOperationException("Central ClusterClient not registered"))); - return; - } - - _log.Debug( - "Forwarding ReconcileSiteRequest for site {0} node {1} ({2} local instance(s)) to central", - msg.SiteIdentifier, msg.NodeId, msg.LocalNameToRevisionHash.Count); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); - - // Internal: send heartbeat tick + // Internal: send heartbeat tick. Receive(_ => SendHeartbeatToCentral()); - // Internal: forward health report to central. The original Sender (the - // AkkaHealthReportTransport's Ask) is forwarded as the ClusterClient.Send - // sender so the central SiteHealthReportAck routes straight back to the - // waiting Ask — making report delivery observable end-to-end (review 01 - // [Medium]). Mirrors the NotificationSubmit ack pattern above. - Receive(msg => - { - if (_centralClient == null) - { - // No ClusterClient registered yet. A non-accepted ack makes the - // sender's counter-restore path treat this tick as a loss. - _log.Warning( - "Cannot forward SiteHealthReport #{0} — no central ClusterClient registered", - msg.SequenceNumber); - Sender.Tell(new SiteHealthReportAck( - msg.SiteId, msg.SequenceNumber, Accepted: false, - Error: "Central ClusterClient not registered")); - return; - } - - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", msg), Sender); - }); - + // Internal: forward the periodic health report (health transport's Ask → ack back), so a + // lost report is observable end-to-end and the sender can restore its per-interval counters. + Receive(msg => _transport.ReportSiteHealth(msg, Sender)); } /// @@ -443,6 +334,12 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers /// protected override void PreStart() { + // Finalize the transport now that the actor context (and _log) exist. The default Akka + // transport is given this actor's logging adapter so the "no ClusterClient registered" + // warnings are preserved exactly. PreStart always runs before any message, so the Receive + // closures above see a non-null _transport. + _transport = _injectedTransport ?? new AkkaCentralTransport(_log); + _log.Info("SiteCommunicationActor started for site {0}", _siteId); // Schedule periodic heartbeat to central. Uses the application heartbeat @@ -478,9 +375,6 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers private void SendHeartbeatToCentral() { - if (_centralClient == null) - return; - var hostname = Environment.MachineName; // Stamp HeartbeatMessage.IsActive with this node's @@ -512,8 +406,17 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers IsActive: isActive, DateTimeOffset.UtcNow); - _centralClient.Tell( - new ClusterClient.Send("/user/central-communication", heartbeat), Self); + // Fire-and-forget on both transports: a failure here must never fault the heartbeat timer + // path. Both real transports swallow their own errors; this catch is a belt-and-braces + // guarantee that no transport (including a future one) can turn a heartbeat into a fault. + try + { + _transport.SendHeartbeat(heartbeat, Self); + } + catch (Exception ex) + { + _log.Debug(ex, "Heartbeat send for site {0} failed; swallowed (heartbeats are fire-and-forget)", _siteId); + } } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index cf1ff099..410b3fda 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -1,11 +1,44 @@ 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. +/// +public enum CentralTransportMode +{ + /// Akka ClusterClient — the transport in production today, and the default. + Akka = 0, + + /// gRPC dial of the central CentralControlService (Phase 1A migration target). + Grpc = 1, +} + /// /// Configuration options for central-site communication, including per-pattern /// timeouts and transport heartbeat settings. /// public class CommunicationOptions { + /// + /// Which transport carries the site→central control messages. Default + /// (ClusterClient) — coexistence rule: a node flips to gRPC only by setting this to Grpc, + /// and rollback is flipping it back. Selecting Grpc requires . + /// + public CentralTransportMode CentralTransport { get; set; } = CentralTransportMode.Akka; + + /// + /// Central control-plane gRPC endpoints (preferred first), e.g. + /// ["http://scadabridge-central-a:8083", "http://scadabridge-central-b:8083"]. Dialled by + /// with sticky failover/failback. Required when + /// is , ignored otherwise. + /// + /// + /// Sites reach central by container/host name, NOT via Traefik (which is HTTP/1 only; gRPC is + /// h2c on the central node's dedicated CentralGrpcPort). + /// + public List CentralGrpcEndpoints { get; set; } = new(); + /// 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/CommunicationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs index ef43f354..1297a9d7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs @@ -66,6 +66,19 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase 0, $"ScadaBridge:Communication:GrpcMaxConcurrentStreams must be positive (was {options.GrpcMaxConcurrentStreams})."); + // The gRPC site→central transport needs at least one central endpoint to dial. Only + // enforced when that transport is selected — the default Akka path ignores the list, so a + // node on ClusterClient must not be forced to declare gRPC endpoints it never uses. + if (options.CentralTransport == CentralTransportMode.Grpc) + { + builder.RequireThat( + options.CentralGrpcEndpoints.Count > 0 + && options.CentralGrpcEndpoints.All(e => !string.IsNullOrWhiteSpace(e)), + "ScadaBridge:Communication:CentralGrpcEndpoints must list at least one non-empty " + + "central gRPC endpoint when CentralTransport is Grpc " + + $"(was {options.CentralGrpcEndpoints.Count} entr{(options.CentralGrpcEndpoints.Count == 1 ? "y" : "ies")})."); + } + // ── Aggregated live alarm cache (plan #10, Task 6) ─────────────────────── // Linger drives a Timer dueTime; TimeSpan.Zero is valid (stop the aggregator // immediately when the last viewer leaves), only a negative value is invalid. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs new file mode 100644 index 00000000..9b8f96fd --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs @@ -0,0 +1,274 @@ +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// A pair (or more) of gRPC channels to the central cluster's control-plane nodes, with the +/// sticky-failover + background-failback policy of the design (§3.5). The site holds one channel +/// per central endpoint, prefers the first, and stays on it until a call proves it unreachable — +/// only then flipping to the next, and only / connect +/// failures count (a never flips or retries, because the +/// call may have run). +/// +/// +/// +/// Sticky. All calls go to the current channel; a healthy preferred endpoint never +/// ping-pongs. flips to the next endpoint (round-robin) when the +/// caller sees the current one refuse a connection. +/// +/// +/// Failback. While off the preferred endpoint a background probe (a cheap Heartbeat +/// ping — always answered, even by a not-yet-ready node) re-checks the preferred one. On the first +/// success new calls return to it; in-flight calls finish where they are. The probe is +/// event-driven: it arms on a flip and stops the moment we are back on the preferred endpoint, so +/// a steady healthy pair spends no cycles. Its cadence backs off exponentially — 1 s, doubling, +/// capped at 60 s — while the preferred endpoint stays down, so a genuinely dead node is not +/// probed hard. +/// +/// +/// Auth. Every channel carries the site's own preshared key and its +/// x-scadabridge-site identity via — the same +/// insecure-h2c call-credentials shape the streaming client uses. The +/// seam lets a test point a channel at an in-process TestServer; production uses a +/// keepalive-configured . +/// +/// +public sealed class CentralChannelProvider : IDisposable +{ + private static readonly TimeSpan DefaultBackoffBase = TimeSpan.FromSeconds(1); + private static readonly TimeSpan DefaultBackoffCap = TimeSpan.FromSeconds(60); + private static readonly TimeSpan DefaultProbeDeadline = TimeSpan.FromSeconds(5); + + private readonly IReadOnlyList _endpoints; + private readonly GrpcChannel[] _channels; + private readonly CentralControlService.CentralControlServiceClient[] _clients; + private readonly ILogger _logger; + private readonly string _siteId; + private readonly TimeSpan _backoffBase; + private readonly TimeSpan _backoffCap; + private readonly TimeSpan _probeDeadline; + private readonly Timer? _failbackTimer; + private readonly object _gate = new(); + + private volatile int _current; // preferred == 0 + private int _consecutiveProbeFailures; + private bool _disposed; + + /// Creates the provider and opens one channel per endpoint. + /// Central control-plane endpoints, preferred first (index 0). Must be non-empty. + /// Resolves this site's preshared key (site-side: a single-key provider). + /// This site's identity, sent as the x-scadabridge-site header. + /// Communication options supplying gRPC keepalive settings. + /// Logger for flip/failback diagnostics. + /// Test seam: per-endpoint ; null uses a production socket handler. + /// Deadline for a failback probe. Null uses 5 s. + /// Initial failback-probe backoff. Null uses 1 s. + /// Maximum failback-probe backoff. Null uses 60 s. + public CentralChannelProvider( + IReadOnlyList endpoints, + ISitePskProvider pskProvider, + string siteId, + CommunicationOptions options, + ILogger logger, + Func? handlerFactory = null, + TimeSpan? probeDeadline = null, + TimeSpan? backoffBase = null, + TimeSpan? backoffCap = null) + { + ArgumentNullException.ThrowIfNull(endpoints); + ArgumentNullException.ThrowIfNull(pskProvider); + ArgumentException.ThrowIfNullOrWhiteSpace(siteId); + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(logger); + if (endpoints.Count == 0) + { + throw new ArgumentException("At least one central gRPC endpoint is required.", nameof(endpoints)); + } + + _endpoints = endpoints; + _logger = logger; + _siteId = siteId; + _backoffBase = backoffBase ?? DefaultBackoffBase; + _backoffCap = backoffCap ?? DefaultBackoffCap; + _probeDeadline = probeDeadline ?? DefaultProbeDeadline; + + _channels = new GrpcChannel[endpoints.Count]; + _clients = new CentralControlService.CentralControlServiceClient[endpoints.Count]; + for (var i = 0; i < endpoints.Count; i++) + { + var channelOptions = new GrpcChannelOptions + { + HttpHandler = handlerFactory?.Invoke(endpoints[i]) ?? new SocketsHttpHandler + { + KeepAlivePingDelay = options.GrpcKeepAlivePingDelay, + KeepAlivePingTimeout = options.GrpcKeepAlivePingTimeout, + KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always, + EnableMultipleHttp2Connections = true, + }, + }.WithSiteCredentials(pskProvider, siteId); + + _channels[i] = GrpcChannel.ForAddress(endpoints[i], channelOptions); + _clients[i] = new CentralControlService.CentralControlServiceClient(_channels[i]); + } + + // Only a multi-endpoint pair can ever fail over, so a lone endpoint needs no probe. + if (endpoints.Count > 1) + { + _failbackTimer = new Timer(_ => _ = FailbackTickAsync(), null, Timeout.Infinite, Timeout.Infinite); + } + } + + /// The number of endpoints in the pair. + public int EndpointCount => _endpoints.Count; + + /// The index of the endpoint calls are currently routed to (preferred == 0). + public int CurrentIndex => _current; + + /// The endpoint address calls are currently routed to. + public string CurrentEndpoint => _endpoints[_current]; + + /// + /// The endpoint index and client calls should use right now. Captured together so a caller can + /// tell exactly which endpoint failed even if a concurrent flip + /// has already moved . + /// + /// The current endpoint index and its client. + public (int Index, CentralControlService.CentralControlServiceClient Client) Current() + { + var idx = _current; + return (idx, _clients[idx]); + } + + /// + /// Reports that the endpoint at refused a connection (an + /// / connect failure). If it is still the current endpoint + /// and another exists, flips to the next one and — when now off the preferred endpoint — arms + /// the failback probe. Idempotent under a concurrent flip: a stale index is ignored. + /// + /// The endpoint index the caller's failed call used. + public void ReportUnavailable(int failedIndex) + { + if (_endpoints.Count < 2) + { + return; // nothing to fail over to + } + + lock (_gate) + { + if (_disposed || failedIndex != _current) + { + return; // a concurrent flip already moved us; do not double-flip + } + + var next = (failedIndex + 1) % _endpoints.Count; + _current = next; + _logger.LogWarning( + "Central control-plane endpoint {Failed} is unavailable; site {SiteId} failed over to {Next}.", + _endpoints[failedIndex], _siteId, _endpoints[next]); + + if (_current != 0) + { + _consecutiveProbeFailures = 0; + ArmFailback(_backoffBase); + } + } + } + + private void ArmFailback(TimeSpan due) + { + if (_disposed) + { + return; + } + + _failbackTimer?.Change(due, Timeout.InfiniteTimeSpan); + } + + private async Task FailbackTickAsync() + { + int currentAtTick = _current; + if (_disposed || currentAtTick == 0) + { + return; // already back on the preferred endpoint (or shutting down) + } + + var preferred = _clients[0]; + try + { + await preferred.HeartbeatAsync( + new HeartbeatDto + { + SiteId = _siteId, + NodeHostname = "failback-probe", + IsActive = false, + Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + deadline: DateTime.UtcNow.Add(_probeDeadline)).ConfigureAwait(false); + + // The preferred endpoint answered — return new calls to it. + lock (_gate) + { + if (_disposed) + { + return; + } + + _current = 0; + _consecutiveProbeFailures = 0; + } + + _logger.LogInformation( + "Central control-plane preferred endpoint {Preferred} is reachable again; site {SiteId} failed back.", + _endpoints[0], _siteId); + } + catch (Exception ex) + { + lock (_gate) + { + if (_disposed || _current == 0) + { + return; + } + + _consecutiveProbeFailures++; + var backoff = NextBackoff(_consecutiveProbeFailures); + _logger.LogDebug(ex, + "Failback probe of preferred central endpoint {Preferred} failed; re-probing in {Backoff}.", + _endpoints[0], backoff); + ArmFailback(backoff); + } + } + } + + private TimeSpan NextBackoff(int failures) + { + // 1 s, doubling, capped at 60 s. Guard the shift against overflow for a long outage. + var exponent = Math.Min(failures - 1, 20); + var scaled = _backoffBase.Ticks * (1L << exponent); + var cap = _backoffCap.Ticks; + return TimeSpan.FromTicks(scaled >= cap || scaled < 0 ? cap : scaled); + } + + /// + public void Dispose() + { + lock (_gate) + { + if (_disposed) + { + return; + } + + _disposed = true; + } + + _failbackTimer?.Dispose(); + foreach (var channel in _channels) + { + channel.Dispose(); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs new file mode 100644 index 00000000..5695c7b2 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs @@ -0,0 +1,258 @@ +using Akka.Actor; +using Grpc.Core; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +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.Communication.Actors; +using AkkaStatus = Akka.Actor.Status; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// The that carries the seven site→central sends over gRPC — the +/// migration target for the Akka ClusterClient path. Each method encodes the message with +/// , dials CentralControlService through the sticky +/// , and delivers the decoded reply (or a transient-failure +/// signal) to the waiting Ask. +/// +/// +/// +/// The actor handlers are synchronous; each method here kicks off the RPC as a detached task and +/// Tells the result to when it completes — IActorRef.Tell +/// is thread-safe, so the reply lands at the Ask exactly as central's ClusterClient reply did. On +/// any non-OK status the reply is , which the S&F / audit / health +/// layers already treat as transient. +/// +/// +/// Cross-node retry only on provably-unsent failures. An +/// (connection refused / node not ready) flips the channel pair and retries once on the peer. +/// A is NEVER retried across nodes — a deploy / write / +/// failover may already have executed, and duplicating it is worse than surfacing a transient +/// failure the layer above tolerates. +/// +/// +/// Per-call deadlines mirror today's Ask timeouts. Notification submit/status → +/// NotificationForwardTimeout (30 s, the value the S&F forwarder and the central service +/// both use); health → HealthReportTimeout (10 s); reconcile → QueryTimeout (30 s); +/// both ingest RPCs → (the one shared 30 s +/// constant). The heartbeat, fire-and-forget with no server-side Ask, is merely bounded by +/// HealthReportTimeout and its failures are swallowed. +/// +/// +public sealed class GrpcCentralTransport : ICentralTransport +{ + private readonly CentralChannelProvider _channels; + private readonly CommunicationOptions _options; + private readonly ILogger _logger; + + /// Creates the transport over a channel pair. + /// The sticky central channel pair. + /// Communication options supplying the per-call deadlines. + /// Logger for failover/fault diagnostics. + public GrpcCentralTransport( + CentralChannelProvider channels, + CommunicationOptions options, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(channels); + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(logger); + + _channels = channels; + _options = options; + _logger = logger; + } + + /// + public void SubmitNotification(NotificationSubmit message, IActorRef replyTo) + { + var dto = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, _options.NotificationForwardTimeout, + (c, o) => c.SubmitNotificationAsync(dto, o), + ack => CentralControlDtoMapper.FromDto(ack)); + } + + /// + public void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo) + { + var dto = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, _options.NotificationForwardTimeout, + (c, o) => c.QueryNotificationStatusAsync(dto, o), + response => CentralControlDtoMapper.FromDto(response)); + } + + /// + public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo) + { + var batch = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout, + (c, o) => c.IngestAuditEventsAsync(batch, o), + ack => new IngestAuditEventsReply(CentralControlDtoMapper.FromIngestAck(ack))); + } + + /// + public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo) + { + var batch = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout, + (c, o) => c.IngestCachedTelemetryAsync(batch, o), + ack => new IngestCachedTelemetryReply(CentralControlDtoMapper.FromIngestAck(ack))); + } + + /// + public void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo) + { + var dto = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, _options.QueryTimeout, + (c, o) => c.ReconcileSiteAsync(dto, o), + response => CentralControlDtoMapper.FromDto(response)); + } + + /// + public void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo) + { + var dto = CentralControlDtoMapper.ToDto(message); + Dispatch(replyTo, _options.HealthReportTimeout, + (c, o) => c.ReportSiteHealthAsync(dto, o), + ack => CentralControlDtoMapper.FromDto(ack)); + } + + /// + public void SendHeartbeat(HeartbeatMessage message, IActorRef self) + { + _ = SendHeartbeatAsync(message); + } + + private async Task SendHeartbeatAsync(HeartbeatMessage message) + { + var (index, client) = _channels.Current(); + var dto = CentralControlDtoMapper.ToDto(message); + try + { + var options = new CallOptions(deadline: DateTime.UtcNow.Add(_options.HealthReportTimeout)); + using var call = client.HeartbeatAsync(dto, options); + await call.ResponseAsync.ConfigureAwait(false); + } + catch (RpcException ex) when (IsConnectFailure(ex)) + { + // Nudge the pair so the next call tries the peer, but never fault: a heartbeat + // failure must not surface on the site's heartbeat timer path. + _channels.ReportUnavailable(index); + _logger.LogDebug(ex, "Heartbeat to central endpoint {Endpoint} was unavailable.", CurrentEndpointSafe()); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Heartbeat to central failed (swallowed — heartbeats are fire-and-forget)."); + } + } + + /// + /// Runs a unary RPC on the current channel, delivers the decoded reply to + /// , and applies the sticky-failover / no-retry-on-deadline policy. + /// + private void Dispatch( + IActorRef replyTo, + TimeSpan timeout, + Func> call, + Func decode) + { + _ = DispatchAsync(replyTo, timeout, call, decode); + } + + private async Task DispatchAsync( + IActorRef replyTo, + TimeSpan timeout, + Func> call, + Func decode) + { + var (index, client) = _channels.Current(); + try + { + var reply = await InvokeAsync(client, timeout, call).ConfigureAwait(false); + replyTo.Tell(decode(reply)); + } + catch (RpcException ex) when (IsConnectFailure(ex)) + { + // Provably unsent: the connection was refused / the node was not ready. Fail over + // to the peer and retry ONCE. This is the only status we retry across nodes. + _channels.ReportUnavailable(index); + var (retryIndex, retryClient) = _channels.Current(); + if (retryIndex != index) + { + try + { + var reply = await InvokeAsync(retryClient, timeout, call).ConfigureAwait(false); + replyTo.Tell(decode(reply)); + return; + } + catch (Exception retryEx) + { + _logger.LogWarning(retryEx, + "Central control-plane call failed on both endpoints; surfacing as transient."); + replyTo.Tell(new AkkaStatus.Failure(retryEx)); + return; + } + } + + replyTo.Tell(new AkkaStatus.Failure(ex)); + } + catch (Exception ex) + { + // DeadlineExceeded / Internal / PermissionDenied / a PSK-resolution throw — do NOT + // retry across nodes (the call may have run). Surface as the transient failure the + // layer above already tolerates. + replyTo.Tell(new AkkaStatus.Failure(ex)); + } + } + + private static async Task InvokeAsync( + CentralControlService.CentralControlServiceClient client, + TimeSpan timeout, + Func> call) + { + var options = new CallOptions(deadline: DateTime.UtcNow.Add(timeout)); + using var asyncCall = call(client, options); + return await asyncCall.ResponseAsync.ConfigureAwait(false); + } + + /// + /// A failure that provably never reached a server — the only class safe to retry on the peer. + /// is deliberately excluded (the call may have run). + /// + /// + /// Two shapes qualify: a server-signalled (e.g. a node + /// that returns Unavailable while it is still starting), and a client-side failure to even + /// start the call — Grpc.Net surfaces a refused/failed connection as + /// "Error starting gRPC call" with the transport exception + /// attached, and there the request never left the client. Anything else — including a deadline, + /// a permission denial, or a generic server-side Internal after the call reached the server — + /// is NOT retried across nodes. + /// + private static bool IsConnectFailure(RpcException ex) + { + if (ex.StatusCode == StatusCode.Unavailable) + { + return true; + } + + // "Error starting gRPC call" is Grpc.Net's marker for a call that could not be sent — a + // refused/failed connection carrying an HttpRequestException. Provably unsent. + return ex.StatusCode == StatusCode.Internal + && (ex.Status.DebugException is HttpRequestException + || ex.Status.Detail.StartsWith("Error starting gRPC call", StringComparison.Ordinal)); + } + + private string CurrentEndpointSafe() + { + try + { + return _channels.CurrentEndpoint; + } + catch + { + return "(unknown)"; + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/StaticSitePskProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/StaticSitePskProvider.cs new file mode 100644 index 00000000..f436ca35 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/StaticSitePskProvider.cs @@ -0,0 +1,43 @@ +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// Site-side over a single fixed key — the one key a site node +/// presents on every control-plane call it makes to central (CommunicationOptions.GrpcPsk). +/// Central's provider resolves a key per site; a site has exactly one, so it ignores the +/// requested siteId and returns its own key. +/// +/// +/// Fail-closed. An empty key throws, matching the contract on +/// and the interceptor's own posture: a node shipped without a key must not degrade to an +/// unauthenticated dial. +/// +public sealed class StaticSitePskProvider : ISitePskProvider +{ + private readonly string _key; + + /// Creates the provider bound to a site's own preshared key. + /// The site's GrpcPsk. Empty is permitted at construction but throws on use. + public StaticSitePskProvider(string key) + { + _key = key ?? string.Empty; + } + + /// + public ValueTask GetAsync(string siteId, CancellationToken ct) + { + if (string.IsNullOrEmpty(_key)) + { + throw new InvalidOperationException( + "No gRPC preshared key is configured for this site (ScadaBridge:Communication:GrpcPsk). " + + "The control plane is fail-closed: an unauthenticated dial does not happen."); + } + + return new ValueTask(_key); + } + + /// + public void Invalidate(string siteId) + { + // A single static key never changes for the process lifetime; nothing to drop. + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 7d58cc63..3624b5ac 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure; using ZB.MOM.WW.ScadaBridge.Communication; using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; using ZB.MOM.WW.ScadaBridge.Host.Actors; using ZB.MOM.WW.ScadaBridge.SiteRuntime; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; @@ -832,13 +833,45 @@ akka {{ _logger, role: siteRole); var dmProxy = dm.Proxy; + // Select the site→central transport behind the coexistence flag (default Akka + // ClusterClient). When gRPC is chosen the site dials CentralControlService directly with + // a sticky-failover channel pair, presenting its own preshared key; the ClusterClient + // below is then not created at all. + ICentralTransport? centralTransport = null; + if (_communicationOptions.CentralTransport == CentralTransportMode.Grpc) + { + var loggerFactory = _serviceProvider.GetRequiredService(); + var channelProvider = new CentralChannelProvider( + _communicationOptions.CentralGrpcEndpoints, + new StaticSitePskProvider(_communicationOptions.GrpcPsk), + _nodeOptions.SiteId!, + _communicationOptions, + loggerFactory.CreateLogger()); + _trackedDisposables.Add(channelProvider); + centralTransport = new GrpcCentralTransport( + channelProvider, + _communicationOptions, + loggerFactory.CreateLogger()); + _logger.LogInformation( + "Site→central transport: gRPC to {Count} central endpoint(s) for site {SiteId}", + _communicationOptions.CentralGrpcEndpoints.Count, _nodeOptions.SiteId); + } + else + { + _logger.LogInformation( + "Site→central transport: Akka ClusterClient (default) for site {SiteId}", + _nodeOptions.SiteId); + } + // Create SiteCommunicationActor for receiving messages from central var siteCommActor = _actorSystem.ActorOf( Props.Create(() => new SiteCommunicationActor( _nodeOptions.SiteId!, _communicationOptions, dmProxy, - activeNodeCheck)), + activeNodeCheck, + null, + centralTransport)), "site-communication"); // Register local handlers with SiteCommunicationActor @@ -957,8 +990,12 @@ akka {{ "Site actors registered. DeploymentManager singleton scoped to role={SiteRole}, SiteCommunicationActor created.", siteRole); - // Create ClusterClient to central if contact points are configured - if (_communicationOptions.CentralContactPoints.Count > 0) + // Create ClusterClient to central if contact points are configured — but only on the Akka + // transport. On the gRPC transport the SiteCommunicationActor already holds a + // GrpcCentralTransport and never receives RegisterCentralClient, so a ClusterClient here + // would be dead weight (and keep an unwanted cross-cluster Akka association alive). + if (_communicationOptions.CentralTransport == CentralTransportMode.Akka + && _communicationOptions.CentralContactPoints.Count > 0) { var contacts = _communicationOptions.CentralContactPoints .Select(cp => ActorPath.Parse($"{cp}/system/receptionist")) diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/AkkaCentralTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/AkkaCentralTransportTests.cs new file mode 100644 index 00000000..6f73cbab --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/AkkaCentralTransportTests.cs @@ -0,0 +1,67 @@ +using Akka.Actor; +using Akka.Cluster.Tools.Client; +using Akka.TestKit.Xunit2; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Actors; + +/// +/// T1A.3: the regression-prone bit of the Akka transport — that a forwarded +/// carries the caller's sender, so central's reply routes +/// straight back to the waiting Ask rather than to the site communication actor — plus the +/// no-ClusterClient-yet fallback that keeps the S&F layer treating the send as transient. +/// +public class AkkaCentralTransportTests : TestKit +{ + [Fact] + public void SubmitNotification_ForwardsToClusterClient_WithReplyToAsSender() + { + var transport = new AkkaCentralTransport(); + var clusterClient = CreateTestProbe(); + var replyTo = CreateTestProbe(); + transport.SetCentralClient(clusterClient.Ref); + + var submit = new NotificationSubmit( + "notif-1", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow); + transport.SubmitNotification(submit, replyTo.Ref); + + // The ClusterClient receives a Send addressed to the central actor... + var send = clusterClient.ExpectMsg(); + Assert.Equal("/user/central-communication", send.Path); + Assert.IsType(send.Message); + + // ...and replying to it lands at replyTo, proving the sender was forwarded (not the + // transport / actor). This is the routing the waiting Ask relies on. + clusterClient.Reply(new NotificationSubmitAck("notif-1", Accepted: true, Error: null)); + replyTo.ExpectMsg(ack => ack.NotificationId == "notif-1" && ack.Accepted); + } + + [Fact] + public void SubmitNotification_WithNoClusterClient_RepliesNonAcceptedToReplyTo() + { + var transport = new AkkaCentralTransport(); + var replyTo = CreateTestProbe(); + + transport.SubmitNotification( + new NotificationSubmit("notif-2", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow), + replyTo.Ref); + + replyTo.ExpectMsg(ack => ack.NotificationId == "notif-2" && !ack.Accepted); + } + + [Fact] + public void IngestAuditEvents_WithNoClusterClient_FaultsTheReplyTo() + { + // The audit drain treats a faulted Ask as transient and keeps rows Pending — so the + // no-client path must be a Status.Failure, not a silent drop. + var transport = new AkkaCentralTransport(); + var replyTo = CreateTestProbe(); + + transport.IngestAuditEvents( + new Commons.Messages.Audit.IngestAuditEventsCommand(new List()), + replyTo.Ref); + + replyTo.ExpectMsg(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/SiteCommunicationActorTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/SiteCommunicationActorTransportTests.cs new file mode 100644 index 00000000..0432a702 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/SiteCommunicationActorTransportTests.cs @@ -0,0 +1,166 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +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.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Actors; + +/// +/// T1A.3: the site communication actor delegates each of the seven site→central sends to the +/// injected , preserving the current Sender as the reply +/// target — and a transport failure surfaces to that sender exactly as the S&F / audit / health +/// layers already expect, while a heartbeat transport fault never faults the actor. +/// +public class SiteCommunicationActorTransportTests : TestKit +{ + private readonly CommunicationOptions _options = new(); + + private (IActorRef actor, ICentralTransport transport) NewActor() + { + var transport = Substitute.For(); + var dmProbe = CreateTestProbe(); + var actor = Sys.ActorOf(Props.Create(() => + new SiteCommunicationActor("site1", _options, dmProbe.Ref, () => false, null, transport))); + return (actor, transport); + } + + [Fact] + public void NotificationSubmit_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + var submit = new NotificationSubmit( + "notif-1", "Operators", "Subj", "Body", "site1", "inst1", "alarmScript", DateTimeOffset.UtcNow); + + actor.Tell(submit, TestActor); + + AwaitAssert(() => transport.Received(1).SubmitNotification( + Arg.Is(m => m.NotificationId == "notif-1"), Arg.Is(TestActor))); + } + + [Fact] + public void NotificationStatusQuery_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + + actor.Tell(new NotificationStatusQuery("corr-1", "notif-1"), TestActor); + + AwaitAssert(() => transport.Received(1).QueryNotificationStatus( + Arg.Is(m => m.NotificationId == "notif-1"), Arg.Is(TestActor))); + } + + [Fact] + public void IngestAuditEvents_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + + actor.Tell(new IngestAuditEventsCommand(new List()), TestActor); + + AwaitAssert(() => transport.Received(1).IngestAuditEvents(Arg.Any(), Arg.Is(TestActor))); + } + + [Fact] + public void IngestCachedTelemetry_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + + actor.Tell(new IngestCachedTelemetryCommand(new List()), TestActor); + + AwaitAssert(() => transport.Received(1).IngestCachedTelemetry(Arg.Any(), Arg.Is(TestActor))); + } + + [Fact] + public void ReconcileSite_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + + actor.Tell( + new ReconcileSiteRequest("site1", "node-a", new Dictionary()), TestActor); + + AwaitAssert(() => transport.Received(1).ReconcileSite( + Arg.Is(m => m.NodeId == "node-a"), Arg.Is(TestActor))); + } + + [Fact] + public void ReportSiteHealth_DelegatesToTransport_WithSenderAsReplyTo() + { + var (actor, transport) = NewActor(); + var report = MinimalHealthReport(sequence: 7); + + actor.Tell(report, TestActor); + + AwaitAssert(() => transport.Received(1).ReportSiteHealth( + Arg.Is(m => m.SequenceNumber == 7), Arg.Is(TestActor))); + } + + [Fact] + public void TransportFailureReply_RoutesBackToTheWaitingSender() + { + // The seam preserves the reply-routing the S&F layer depends on: when the transport + // answers the captured replyTo with a Status.Failure (its transient-failure signal on a + // non-OK status), that failure reaches the original sender — here the test actor — so the + // waiting Ask faults, exactly as it did on the ClusterClient path. + var transport = Substitute.For(); + transport + .When(t => t.SubmitNotification(Arg.Any(), Arg.Any())) + .Do(ci => ci.Arg().Tell( + new Status.Failure(new InvalidOperationException("central unavailable")))); + + var dmProbe = CreateTestProbe(); + var actor = Sys.ActorOf(Props.Create(() => + new SiteCommunicationActor("site1", _options, dmProbe.Ref, () => false, null, transport))); + + actor.Tell(new NotificationSubmit( + "notif-x", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow), TestActor); + + var failure = ExpectMsg(); + Assert.IsType(failure.Cause); + } + + [Fact] + public void HeartbeatTransportThrow_DoesNotFaultTheActor() + { + // A transport whose SendHeartbeat throws must not fault the actor — heartbeats are + // fire-and-forget and their failure is swallowed. Prove the actor still serves messages + // after a heartbeat that threw. + var transport = Substitute.For(); + transport + .When(t => t.SendHeartbeat(Arg.Any(), Arg.Any())) + .Do(_ => throw new InvalidOperationException("boom")); + + var dmProbe = CreateTestProbe(); + var actor = Sys.ActorOf(Props.Create(() => + new SiteCommunicationActor( + "site1", + new CommunicationOptions { ApplicationHeartbeatInterval = TimeSpan.FromMilliseconds(50) }, + dmProbe.Ref, () => true, null, transport))); + + // Let the heartbeat timer fire a few times (each throws inside the transport). + Thread.Sleep(250); + + // The actor is still alive and delegating: a subsequent send is handled normally. + actor.Tell(new NotificationSubmit( + "after-heartbeat", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow), TestActor); + AwaitAssert(() => transport.Received(1).SubmitNotification( + Arg.Is(m => m.NotificationId == "after-heartbeat"), Arg.Is(TestActor))); + } + + private static SiteHealthReport MinimalHealthReport(long sequence) => new( + SiteId: "site1", + SequenceNumber: sequence, + ReportTimestamp: DateTimeOffset.UtcNow, + DataConnectionStatuses: new Dictionary(), + TagResolutionCounts: new Dictionary(), + ScriptErrorCount: 0, + AlarmEvaluationErrorCount: 0, + StoreAndForwardBufferDepths: new Dictionary(), + DeadLetterCount: 0, + DeployedInstanceCount: 0, + EnabledInstanceCount: 0, + DisabledInstanceCount: 0); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs index dafba565..441f62f9 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs @@ -104,4 +104,57 @@ public class CommunicationOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage); } + + // ── T1A.3: gRPC central transport endpoints (required only when selected) ──── + + [Fact] + public void GrpcTransport_WithNoEndpoints_IsRejected() + { + var result = Validate(new CommunicationOptions + { + CentralTransport = CentralTransportMode.Grpc, + CentralGrpcEndpoints = new List(), + }); + Assert.True(result.Failed); + Assert.Contains("CentralGrpcEndpoints", result.FailureMessage); + } + + [Fact] + public void GrpcTransport_WithBlankEndpoint_IsRejected() + { + var result = Validate(new CommunicationOptions + { + CentralTransport = CentralTransportMode.Grpc, + CentralGrpcEndpoints = new List { " " }, + }); + Assert.True(result.Failed); + Assert.Contains("CentralGrpcEndpoints", result.FailureMessage); + } + + [Fact] + public void GrpcTransport_WithEndpoints_IsValid() + { + var result = Validate(new CommunicationOptions + { + CentralTransport = CentralTransportMode.Grpc, + CentralGrpcEndpoints = new List + { + "http://scadabridge-central-a:8083", + "http://scadabridge-central-b:8083", + }, + }); + Assert.True(result.Succeeded, result.FailureMessage); + } + + [Fact] + public void AkkaTransport_IgnoresMissingGrpcEndpoints() + { + // The default transport must not be forced to declare gRPC endpoints it never dials. + var result = Validate(new CommunicationOptions + { + CentralTransport = CentralTransportMode.Akka, + CentralGrpcEndpoints = new List(), + }); + Assert.True(result.Succeeded, result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcCentralTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcCentralTransportTests.cs new file mode 100644 index 00000000..733829c0 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcCentralTransportTests.cs @@ -0,0 +1,373 @@ +using System.Collections.Concurrent; +using Akka.Actor; +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.Notification; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// T1A.3: + over a real +/// gRPC stack (two in-process central nodes, the real +/// and ). Proves +/// the sticky failover/failback policy, the PSK + site-header attachment, the per-call deadline, +/// and — the hard rule — no cross-node retry on DeadlineExceeded. +/// +/// +/// A "down" node is modelled by a that throws before reaching the +/// TestServer, so BOTH the unary call and the failback Heartbeat probe see it as +/// Unavailable — the honest shape of a refused connection, and the only class the transport +/// fails over on. Readiness is always set, so a node that is "up" answers everything. +/// +public class GrpcCentralTransportTests : IAsyncLifetime +{ + private const string SiteA = "site-a"; + private const string SiteAKey = "site-a-preshared-key"; + private const string EndpointA = "http://central-a/"; + private const string EndpointB = "http://central-b/"; + + private ActorSystem _system = null!; + private CentralNode _nodeA = null!; + private CentralNode _nodeB = null!; + + /// + public async Task InitializeAsync() + { + _system = ActorSystem.Create("grpc-central-transport-test"); + _nodeA = await CentralNode.StartAsync(_system, "A", SiteA, SiteAKey, repliesToSubmit: true); + _nodeB = await CentralNode.StartAsync(_system, "B", SiteA, SiteAKey, repliesToSubmit: true); + } + + /// + public async Task DisposeAsync() + { + await _nodeA.DisposeAsync(); + await _nodeB.DisposeAsync(); + await _system.Terminate(); + } + + private CentralChannelProvider NewProvider(string? pskKey = SiteAKey) => new( + new[] { EndpointA, EndpointB }, + new FixedPskProvider(pskKey), + SiteA, + new CommunicationOptions(), + NullLogger.Instance, + handlerFactory: HandlerFor, + probeDeadline: TimeSpan.FromSeconds(2), + backoffBase: TimeSpan.FromMilliseconds(50), + backoffCap: TimeSpan.FromMilliseconds(200)); + + private HttpMessageHandler HandlerFor(string endpoint) => endpoint == EndpointA + ? new ToggleHandler(_nodeA.Server.CreateHandler(), () => _nodeA.IsUp) + : new ToggleHandler(_nodeB.Server.CreateHandler(), () => _nodeB.IsUp); + + private GrpcCentralTransport NewTransport(CentralChannelProvider provider, CommunicationOptions? options = null) + => new(provider, options ?? new CommunicationOptions(), NullLogger.Instance); + + [Fact] + public async Task HappyPath_ReachesThePreferredNode_AndRoutesTheAckBack() + { + using var provider = NewProvider(); + var transport = NewTransport(provider); + var inbox = new Capture(_system); + + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + + var ack = Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.True(ack.Accepted); + Assert.Equal("n1", ack.NotificationId); + Assert.Equal(0, provider.CurrentIndex); // stayed on preferred + Assert.Equal(1, _nodeA.SubmitCount); + Assert.Equal(0, _nodeB.SubmitCount); + } + + [Fact] + public async Task Sticky_StaysOnThePreferredNode_WhileHealthy() + { + using var provider = NewProvider(); + var transport = NewTransport(provider); + + for (var i = 0; i < 4; i++) + { + var inbox = new Capture(_system); + transport.SubmitNotification(NewSubmit($"n{i}"), inbox.Ref); + Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + } + + Assert.Equal(0, provider.CurrentIndex); + Assert.Equal(4, _nodeA.SubmitCount); + Assert.Equal(0, _nodeB.SubmitCount); + } + + [Fact] + public async Task Failover_FlipsToThePeer_WhenThePreferredIsUnavailable() + { + using var provider = NewProvider(); + var transport = NewTransport(provider); + _nodeA.IsUp = false; // preferred refuses connections + + var inbox = new Capture(_system); + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + + var ack = Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.True(ack.Accepted); + Assert.Equal(1, provider.CurrentIndex); // flipped to the peer + Assert.Equal(0, _nodeA.SubmitCount); + Assert.Equal(1, _nodeB.SubmitCount); + } + + [Fact] + public async Task Failback_ReturnsToThePreferred_OnceItIsReachableAgain() + { + using var provider = NewProvider(); + var transport = NewTransport(provider); + + // Take the preferred down and drive one call so we flip to the peer + arm the failback probe. + _nodeA.IsUp = false; + var inbox = new Capture(_system); + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.Equal(1, provider.CurrentIndex); + + // Bring the preferred back; the background probe should fail us back within a few backoffs. + _nodeA.IsUp = true; + await WaitUntil(() => provider.CurrentIndex == 0, TimeSpan.FromSeconds(5)); + Assert.Equal(0, provider.CurrentIndex); + + // New calls resume on the preferred node. + var inbox2 = new Capture(_system); + transport.SubmitNotification(NewSubmit("n2"), inbox2.Ref); + Assert.IsType(inbox2.Receive(TimeSpan.FromSeconds(5))); + Assert.True(_nodeA.SubmitCount >= 1); + } + + [Fact] + public async Task PskAndSiteHeader_AreAttached_SoTheGatedCallReachesTheService() + { + // The service is gated by CentralControlAuthInterceptor; a call that reaches it (and gets + // Accepted) proves both the bearer PSK and the x-scadabridge-site header were attached. + using var provider = NewProvider(pskKey: SiteAKey); + var transport = NewTransport(provider); + var inbox = new Capture(_system); + + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + + var ack = Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.True(ack.Accepted); + } + + [Fact] + public async Task WrongPsk_IsRejected_AndNotRetriedOnThePeer() + { + // PermissionDenied is not a connect failure — the transport surfaces it as a transient + // Status.Failure without flipping to the peer. + using var provider = NewProvider(pskKey: "the-wrong-key"); + var transport = NewTransport(provider); + var inbox = new Capture(_system); + + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + + Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.Equal(0, provider.CurrentIndex); // no flip + Assert.Equal(0, _nodeB.SubmitCount); // peer never tried + } + + [Fact] + public async Task DeadlineExceeded_IsNotRetriedOnThePeer() + { + // THE hard rule. Node A is UP but never replies, so the call deadlines. The transport must + // surface Status.Failure and must NOT try node B (the call may already have executed). + _nodeA.SetBlackHole(); + var shortDeadline = new CommunicationOptions { NotificationForwardTimeout = TimeSpan.FromMilliseconds(300) }; + + using var provider = NewProvider(); + var transport = NewTransport(provider, shortDeadline); + var inbox = new Capture(_system); + + transport.SubmitNotification(NewSubmit("n1"), inbox.Ref); + + // A per-call deadline is applied (the call returns fast instead of hanging on the black hole). + Assert.IsType(inbox.Receive(TimeSpan.FromSeconds(5))); + Assert.Equal(0, provider.CurrentIndex); // no failover on a deadline + Assert.Equal(0, _nodeB.SubmitCount); // peer never tried + } + + private static NotificationSubmit NewSubmit(string id) => new( + NotificationId: id, + ListName: "ops", + Subject: "s", + Body: "b", + SourceSiteId: SiteA, + SourceInstanceId: null, + SourceScript: null, + SiteEnqueuedAt: DateTimeOffset.UtcNow); + + private static async Task WaitUntil(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return; + } + + await Task.Delay(25); + } + } + + /// + /// A raw message sink used as the transport's replyTo. Unlike Akka's Inbox, which + /// rethrows a 's cause on receive, this captures every message + /// verbatim so a test can assert on the itself. + /// + private sealed class Capture + { + private readonly BlockingCollection _messages = new(); + + public Capture(ActorSystem system) + { + Ref = system.ActorOf(Props.Create(() => new CaptureActor(_messages))); + } + + public IActorRef Ref { get; } + + public object Receive(TimeSpan timeout) + => _messages.TryTake(out var message, timeout) + ? message + : throw new TimeoutException("No message captured within the timeout."); + + private sealed class CaptureActor : ReceiveActor + { + public CaptureActor(BlockingCollection messages) => ReceiveAny(messages.Add); + } + } + + /// A gRPC channel handler that throws (a refused connection) while its node is "down". + private sealed class ToggleHandler : DelegatingHandler + { + private readonly Func _isUp; + + public ToggleHandler(HttpMessageHandler inner, Func isUp) + { + InnerHandler = inner; + _isUp = isUp; + } + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + if (!_isUp()) + { + throw new HttpRequestException("simulated central node down"); + } + + return await base.SendAsync(request, cancellationToken).ConfigureAwait(false); + } + } + + private sealed class FixedPskProvider(string? key) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) + => key is null ? throw new InvalidOperationException("no key") : new ValueTask(key); + + public void Invalidate(string siteId) { } + } + + /// One in-process central node: TestServer + real service/interceptor + a stub actor. + private sealed class CentralNode : IAsyncDisposable + { + private IHost _host = null!; + private IActorRef _stub = null!; + private readonly StubCounters _counters = new(); + + public TestServer Server { get; private set; } = null!; + public volatile bool IsUp = true; + public int SubmitCount => _counters.Submits; + + public static async Task StartAsync( + ActorSystem system, string label, string site, string key, bool repliesToSubmit) + { + var node = new CentralNode(); + node._stub = system.ActorOf( + Props.Create(() => new StubCentralActor(node._counters, repliesToSubmit)), $"stub-{label}"); + + var service = new CentralControlGrpcService( + NullLogger.Instance, + Options.Create(new CommunicationOptions())); + service.SetReady(node._stub); + + var psk = new MapPskProvider(new Dictionary { [site] = key }); + + node._host = await new HostBuilder() + .ConfigureWebHost(web => web + .UseTestServer() + .ConfigureServices(services => + { + services.AddGrpc(o => o.Interceptors.Add()); + services.AddSingleton(psk); + services.AddSingleton(service); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapGrpcService()); + })) + .StartAsync(); + + node.Server = node._host.GetTestServer(); + return node; + } + + /// Switches the node's actor to a black hole that counts but never replies. + public void SetBlackHole() => _counters.BlackHole = true; + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + private sealed class StubCounters + { + private int _submits; + public int Submits => Volatile.Read(ref _submits); + public void IncrementSubmits() => Interlocked.Increment(ref _submits); + public volatile bool BlackHole; + } + + private sealed class StubCentralActor : ReceiveActor + { + public StubCentralActor(StubCounters counters, bool repliesToSubmit) + { + Receive(msg => + { + counters.IncrementSubmits(); + if (repliesToSubmit && !counters.BlackHole) + { + Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: true, Error: null)); + } + }); + + // Heartbeat lands here as a Tell (the failback probe); ignore it, no reply expected. + ReceiveAny(_ => { }); + } + } + + private sealed class MapPskProvider(IReadOnlyDictionary keys) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) + => keys.TryGetValue(siteId, out var key) + ? new ValueTask(key) + : throw new InvalidOperationException($"no key for '{siteId}'"); + + public void Invalidate(string siteId) { } + } + } +}