using Akka.Actor; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; /// /// Runs on EVERY site node (NOT a singleton) so a standby that was DOWN during a deploy /// self-heals on its next restart. On startup the actor performs one best-effort /// reconciliation pass: /// /// read the node's local deployed inventory from SQLite, /// report it to central via the SiteCommunicationActor Ask /// ( → ), /// fetch each gap item's config over HTTP and guarded-write it, /// ask the Deployment Manager singleton to START each enabled instance it just /// healed, and /// LOG (never delete) any orphan the node still has but central no longer deploys. /// /// /// /// /// Best-effort throughout. A central-unreachable / timed-out Ask is caught, logged /// at Warning, and the pass simply ends — reconcile re-runs on the next node startup; it is a /// self-heal, not a critical path. A per-item fetch/write failure is caught and logged, then /// the remaining gap items continue (one bad item must not abort the rest). The actor never /// crashes on these failures. /// /// /// The pass runs after a small startup delay (so the site's central transport has time to /// connect) and is driven entirely off the actor thread: the Ask + fetch + write happen in /// an awaited continuation whose summary is captured in an internal message /// piped back to Self. The actor thread never blocks. /// /// /// The site does NOT carry the central fetch base URL in its own config — it uses /// from central's reply. /// /// /// The pass heals the runtime, not just the file (deferred register row 67). Storing a /// config starts nothing by itself: the Deployment Manager creates Instance Actors from the /// boot-time SQLite load, which has already run by the time this pass fires. So for every /// ENABLED instance whose config this pass stored, an /// is sent to the Deployment Manager singleton, which starts the actor if it is not already /// running. Without it, an instance whose config was missing at boot (lost volume, first boot /// after a restore, rehomed data dir) needed a SECOND restart before it ran again. /// The command is advisory, idempotent, and deliberately fire-and-forget — a site with no /// Deployment Manager ref configured simply keeps the old file-only behavior. /// /// public sealed class SiteReconciliationActor : ReceiveActor, IWithTimers { private const string StartupTimerKey = "reconcile-startup"; private readonly SiteStorageService _storage; private readonly IDeploymentConfigFetcher _configFetcher; private readonly IActorRef _siteCommunicationActor; private readonly IActorRef? _deploymentManager; private readonly string _siteIdentifier; private readonly string _nodeId; private readonly ILogger _logger; private readonly TimeSpan _initialDelay; private readonly TimeSpan _askTimeout; /// Akka timer scheduler injected by the framework via . public ITimerScheduler Timers { get; set; } = null!; /// /// Initializes the per-node startup-reconciliation actor. /// /// Site-local SQLite store — read for the inventory, written for the gap. /// Fetches a deployment's flattened config JSON from central over HTTP. /// /// The site's SiteCommunicationActor; it forwards the /// over the site's central transport and routes /// the back to this actor's Ask. /// /// This node's site identifier (resolved by central). /// This node's semantic id (e.g. node-a/node-b), for logging/diagnostics. /// Logger. /// /// Delay before the single startup pass, giving the site's central transport time to connect. /// Defaults to 5 seconds. /// /// Round-trip timeout for the reconcile Ask to central. Defaults to 30 seconds. /// /// The Deployment Manager singleton PROXY. Every enabled instance this pass heals is /// followed by a fire-and-forget so the runtime /// heals with the file (register row 67). Routing through the proxy — rather than starting /// anything locally — is what keeps Instance Actors under the singleton no matter which /// node's reconcile pass found the gap. Optional: when null the pass stores configs and /// starts nothing, exactly as it did before. /// public SiteReconciliationActor( SiteStorageService storage, IDeploymentConfigFetcher configFetcher, IActorRef siteCommunicationActor, string siteIdentifier, string nodeId, ILogger logger, TimeSpan? initialDelay = null, TimeSpan? askTimeout = null, IActorRef? deploymentManager = null) { _storage = storage; _configFetcher = configFetcher; _siteCommunicationActor = siteCommunicationActor; _deploymentManager = deploymentManager; _siteIdentifier = siteIdentifier; _nodeId = nodeId; _logger = logger; _initialDelay = initialDelay ?? TimeSpan.FromSeconds(5); _askTimeout = askTimeout ?? TimeSpan.FromSeconds(30); Receive(_ => RunReconcilePassAsync().PipeTo(Self)); Receive(HandleReconcilePassResult); // Defensive: RunReconcilePassAsync is designed never to throw (it returns a faulted // ReconcilePassResult instead), but if anything unexpected faults the piped Task the // Status.Failure would otherwise go to dead letters silently. Log it instead. Receive(f => _logger.LogWarning(f.Cause, "Reconcile pass faulted unexpectedly for site {Site} node {Node}", _siteIdentifier, _nodeId)); } /// protected override void PreStart() { base.PreStart(); // One-shot pass after a small delay so the site's central transport can connect first. // Non-blocking: the timer fires RunReconcile back onto this actor's mailbox. Timers.StartSingleTimer(StartupTimerKey, RunReconcile.Instance, _initialDelay); _logger.LogInformation( "SiteReconciliationActor started for site {Site} node {Node}; startup reconcile scheduled in {Delay}", _siteIdentifier, _nodeId, _initialDelay); } /// /// Runs the full reconcile pass off the actor thread. Never throws: a central-unreachable /// Ask (or any other top-level failure) is captured as a faulted /// ; per-item fetch/write failures are caught per item so /// the rest of the gap still applies. /// private async Task RunReconcilePassAsync() { Dictionary localMap; try { var configs = await _storage.GetAllDeployedConfigsAsync().ConfigureAwait(false); localMap = new Dictionary(configs.Count, StringComparer.Ordinal); foreach (var c in configs) localMap[c.InstanceUniqueName] = c.RevisionHash; } catch (Exception ex) { return ReconcilePassResult.Faulted(ex); } // Report inventory to central and get fresh fetch tokens for the gap. Best-effort: // a central-unreachable / timed-out Ask faults here and is reported as a faulted pass // (logged Warning; reconcile re-runs next startup). ReconcileSiteResponse response; try { response = await _siteCommunicationActor .Ask( new ReconcileSiteRequest(_siteIdentifier, _nodeId, localMap), _askTimeout) .ConfigureAwait(false); } catch (Exception ex) { return ReconcilePassResult.Faulted(ex); } var fetched = 0; var failed = 0; // Enabled instances healed by this pass, to be started via the Deployment Manager once // the pass reports back on the actor thread (row 67). var healedEnabled = new List(); // Fetch + guarded-write each gap item. Per-item failure must not abort the rest. foreach (var item in response.Gap) { try { var configJson = await _configFetcher .FetchAsync(response.CentralFetchBaseUrl, item.DeploymentId, item.FetchToken, CancellationToken.None) .ConfigureAwait(false); await _storage.StoreDeployedConfigIfNewerAsync( item.InstanceUniqueName, configJson, item.DeploymentId, item.RevisionHash, item.IsEnabled) .ConfigureAwait(false); fetched++; // Only ENABLED instances are candidates to start; a disabled one correctly // keeps a config row with no running actor. The Deployment Manager re-checks // this against the durable row anyway — this is just the cheap filter. if (item.IsEnabled) healedEnabled.Add(item.InstanceUniqueName); _logger.LogInformation( "Reconcile: fetched + stored config for {Instance} (deployment {DeploymentId}, rev {Revision})", item.InstanceUniqueName, item.DeploymentId, item.RevisionHash); } catch (DeploymentConfigFetchException ex) when (ex.IsSuperseded) { // 404 = superseded/expired between staging and fetch; a newer deploy will // replicate it. Not a failure — skip quietly. _logger.LogInformation( "Reconcile: skip {Instance} (deployment {DeploymentId}) — superseded/expired", item.InstanceUniqueName, item.DeploymentId); } catch (Exception ex) { failed++; _logger.LogError(ex, "Reconcile: failed to fetch/store config for {Instance} (deployment {DeploymentId}) — continuing with remaining items", item.InstanceUniqueName, item.DeploymentId); } } // Orphans: present locally but no longer deployed at central. LOG only — never delete // (a stale local row is harmless; deleting risks dropping a config a later deploy needs). foreach (var name in response.OrphanNames) { _logger.LogWarning( "Reconcile: local instance {Instance} is no longer deployed at central — leaving in place; manual cleanup may be needed", name); } return ReconcilePassResult.Completed(fetched, failed, response.OrphanNames.Count, healedEnabled); } private void HandleReconcilePassResult(ReconcilePassResult result) { if (result.Error != null) { // Best-effort: a failed pass (central unreachable, Ask timeout, local read error) is // logged at Warning and the actor stays alive. Reconcile re-runs on the next startup. _logger.LogWarning(result.Error, "Reconcile pass for site {Site} node {Node} did not complete (central unreachable or read error) — will retry on next startup", _siteIdentifier, _nodeId); return; } _logger.LogInformation( "Reconcile pass for site {Site} node {Node} complete: {Fetched} fetched, {Failed} failed, {Orphans} orphan(s)", _siteIdentifier, _nodeId, result.Fetched, result.Failed, result.Orphans); // Heal the RUNTIME, not just the file (row 67). Storing a config starts nothing: the // Deployment Manager's boot-time SQLite load has already run by now, so without this // an instance missing at boot would stay dead until the next restart. Fire-and-forget // through the singleton proxy — the handler is idempotent, re-checks the durable row's // enabled state, and defers to any in-flight deploy. if (_deploymentManager == null || result.HealedEnabled.Count == 0) return; foreach (var instanceName in result.HealedEnabled) { _logger.LogInformation( "Reconcile: requesting start of healed instance {Instance} on site {Site}", instanceName, _siteIdentifier); _deploymentManager.Tell( new EnsureInstanceStartedCommand(instanceName, "startup reconcile")); } } // ── Internal messages ── /// Self-tick that drives the one-shot startup reconcile pass. private sealed class RunReconcile { public static readonly RunReconcile Instance = new(); private RunReconcile() { } } /// /// Summary of one reconcile pass, piped to Self for logging and for the /// ensure-started dispatch. is built off the actor thread but /// only ever read on it, so no synchronization is needed. /// private sealed record ReconcilePassResult( int Fetched, int Failed, int Orphans, IReadOnlyList HealedEnabled, Exception? Error) { /// Creates a result for a reconcile pass that ran to completion. /// Number of instances successfully fetched/reconciled. /// Number of instances that failed to reconcile. /// Number of local instances no longer deployed at central. /// /// Names of the ENABLED instances whose config this pass stored — each gets an /// so the runtime heals with the file. /// /// A completed with no error. public static ReconcilePassResult Completed( int fetched, int failed, int orphans, IReadOnlyList healedEnabled) => new(fetched, failed, orphans, healedEnabled, null); /// Creates a result for a reconcile pass that failed before completing. /// The exception that caused the pass to fail. /// A faulted carrying the error. public static ReconcilePassResult Faulted(Exception error) => new(0, 0, 0, [], error); } }