using Akka.Actor; using Akka.Cluster; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts; 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.Instance; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; using ZB.MOM.WW.ScadaBridge.SiteEventLogging; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; /// /// Site-side Deployment Manager — runs as a cluster singleton within the site cluster. /// On startup, reads all deployed configs from SQLite and creates Instance Actors /// for enabled instances in staggered batches. /// /// Handles: DeployInstance, DisableInstance, EnableInstance, DeleteInstance. /// /// Supervision strategy: OneForOneStrategy with Resume for Instance Actors /// so that a single instance failure does not cascade to siblings. /// public class DeploymentManagerActor : ReceiveActor, IWithTimers { private readonly SiteStorageService _storage; private readonly ScriptCompilationService _compilationService; // Pure, off-thread deploy-time compile gate. A site-side compile failure // must reject the deployment (spec: no partial state applied), so every // DeployInstanceCommand is validated before any actor/persistence work. private readonly DeployCompileValidator _deployCompileValidator; private readonly SharedScriptLibrary _sharedScriptLibrary; private readonly SiteStreamManager? _streamManager; private readonly SiteRuntimeOptions _options; private readonly ILogger _logger; /// /// Shared logger factory used to mint loggers. /// Reused across every /// call rather than newing a per-instance factory that is never disposed. /// When the host injects its configured factory the Instance Actor logs are /// routed through the application's logging providers. /// private readonly ILoggerFactory _loggerFactory; private readonly IActorRef? _dclManager; private readonly IActorRef? _replicationActor; private readonly ISiteHealthCollector? _healthCollector; private readonly IServiceProvider? _serviceProvider; /// /// Notify-and-fetch: fetches a deployment's flattened config from central /// over HTTP when a arrives. Optional — null on /// nodes/tests that never receive a refresh; the active site path supplies it via DI. /// private readonly IDeploymentConfigFetcher? _configFetcher; /// /// Delay before re-attempting the startup deployed-config load after a /// transient failure (S5). Fixed-interval, unlimited retries — consistent with /// the system's fixed-interval retry philosophy. Overridable via the ctor for tests. /// private readonly TimeSpan _startupLoadRetryInterval; /// /// Loads all deployed configs at startup. Defaults to /// ; overridable via the /// ctor so a test can drive a transient load failure (S5). /// private readonly Func>> _configLoader; /// /// Set once the startup deployed-config load has succeeded. Guards against a /// stale re-running the load after success. /// private bool _startupLoadCompleted; private readonly Dictionary _instanceActors = new(); /// /// Tracks Instance Actors that are terminating as part of a redeployment, keyed by /// the terminating actor ref. The buffered command is applied once /// confirms the child has fully stopped. /// private readonly Dictionary _pendingRedeploys = new(); /// /// Name → terminating actor ref shadow of . /// Required because a third for the same instance /// arriving WHILE a redeploy is still mid-termination would otherwise see /// _instanceActors.TryGetValue == false and fall through to /// + , where /// Context.ActorOf(props, instanceName) throws InvalidActorNameException /// — the child name is still registered until the signal fires. /// The supervisor's Stop directive then drops the deploy command silently, leaving the /// deployer waiting forever and persistence dangling. The shadow index lets /// detect the mid-termination state and overwrite the /// buffered pending command (last-write-wins) instead of trying to create a fresh actor. /// Cleared in alongside . /// private readonly Dictionary _terminatingActorsByName = new(); /// /// Authoritative set of deployed instance config names (enabled /// AND disabled). The deployed/disabled health counts are derived from this set's /// size, so add-on-deploy / remove-on-delete keeps the count correct for every /// path — including deleting a DISABLED instance, which has a config row but is /// absent from both and . /// private readonly HashSet _deployedInstanceNames = new(); /// /// Deployments awaiting their persistence result, keyed by instance name (S6). /// The entry is added when the Instance Actor is created optimistically and removed /// by whichever of two paths fires first: /// /// on a normal persistence outcome, /// which then replies Success/Failed; /// the S6 fallback when the actor dies during /// init, which replies Failed and rolls back the optimistic state. /// /// Removing the entry in exactly one of those paths guarantees the deployer receives /// exactly one regardless of ordering. /// private readonly Dictionary _inFlightDeployments = new(); /// /// Instances whose deployment was failed by the S6 Terminated fallback WHILE their /// optimistic StoreDeployedConfigAsync was still in flight. The persisted-row /// rollback must run AFTER the store commits (the two are independent background /// tasks with no ordering guarantee), so the removal is deferred here and executed by /// when the store's result finally lands. /// private readonly HashSet _initFailedPendingRowRemoval = new(); /// Akka timer scheduler injected by the framework via . public ITimerScheduler Timers { get; set; } = null!; /// Initializes the actor, wires all message handlers, and prepares for staggered Instance Actor startup. /// Site SQLite storage service for deployed configuration persistence. /// Script compilation service used when creating Instance Actors. /// Shared script library compiled before Instance Actors are created. /// Optional site-wide stream manager; null in unit tests. /// Site runtime options (startup batch size, stagger interval, etc.). /// Logger instance. /// Optional Data Connection Layer manager actor. /// Optional standby replication actor. /// Optional site health collector. /// Optional service provider for resolving per-instance services. /// Optional logger factory for creating Instance Actor loggers. /// /// Optional notify-and-fetch config fetcher. Required for the /// path; null on nodes/tests that never /// receive a refresh. /// /// Optional retry interval used when loading /// deployed configuration at startup; defaults to 5 seconds when null. /// Optional override for loading all deployed configurations /// at startup; defaults to reading from . Primarily for tests. public DeploymentManagerActor( SiteStorageService storage, ScriptCompilationService compilationService, SharedScriptLibrary sharedScriptLibrary, SiteStreamManager? streamManager, SiteRuntimeOptions options, ILogger logger, IActorRef? dclManager = null, IActorRef? replicationActor = null, ISiteHealthCollector? healthCollector = null, IServiceProvider? serviceProvider = null, ILoggerFactory? loggerFactory = null, IDeploymentConfigFetcher? configFetcher = null, TimeSpan? startupLoadRetryInterval = null, Func>>? configLoader = null) { _storage = storage; _compilationService = compilationService; _deployCompileValidator = new DeployCompileValidator(compilationService); _sharedScriptLibrary = sharedScriptLibrary; _streamManager = streamManager; _options = options; _dclManager = dclManager; _replicationActor = replicationActor; _healthCollector = healthCollector; _serviceProvider = serviceProvider; _configFetcher = configFetcher; _startupLoadRetryInterval = startupLoadRetryInterval ?? TimeSpan.FromSeconds(5); _configLoader = configLoader ?? (() => _storage.GetAllDeployedConfigsAsync()); _logger = logger; // Reuse a single logger factory for all Instance Actors. // Prefer an explicitly injected factory, fall back to one resolved from // the service provider, and only as a last resort use NullLoggerFactory — // never a per-instance `new LoggerFactory()` that leaks undisposed. _loggerFactory = loggerFactory ?? serviceProvider?.GetService(typeof(ILoggerFactory)) as ILoggerFactory ?? Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance; // Lifecycle commands Receive(cmd => HandleDeploy(cmd, Sender)); Receive(HandleDisable); Receive(HandleEnable); Receive(HandleDelete); // Notify-and-fetch: central sends a small RefreshDeploymentCommand; // the active singleton fetches the flattened config over HTTP, then reuses the // existing apply path. The two internal results carry the fetched config (or the // fetch error) back onto the actor thread along with the captured original sender. Receive(HandleRefreshDeployment); Receive(HandleRefreshFetched); Receive(HandleRefreshFetchFailed); // Query-the-site-before-redeploy idempotency. // Central asks for the instance's currently-applied deployment identity // before re-sending a deployment whose prior record is stuck InProgress // or Failed due to a timeout. Receive(HandleDeploymentStateQuery); // Handle system-wide artifact deployment Receive(HandleDeployArtifacts); // Artifact-deploy DCL push, dispatched back from the // off-thread persistence task so the hash-cache mutation stays // actor-thread-confined. Receive(HandleApplyArtifactDataConnectionsToDcl); // Debug View — route to Instance Actors Receive(RouteDebugViewSubscribe); Receive(RouteDebugViewUnsubscribe); Receive(RouteDebugSnapshot); // Inbound API Route.To().Call() — route to Instance Actors Receive(RouteInboundApiCall); Receive(RouteInboundApiGetAttributes); Receive(RouteInboundApiSetAttributes); Receive(RouteInboundApiWaitForAttribute); // OPC UA Tag Browser — singleton-only re-forward to local /user/dcl-manager. // BrowseNodeCommand is routed to this singleton (active node) by // SiteCommunicationActor so the dcl-manager we forward to is guaranteed // to be the one holding the live DataConnectionActor children. ActorSelection // has no Forward() extension in this Akka.NET version, so we Tell with the // original Sender preserved (semantically identical to Forward). Receive(msg => Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender)); // Test Bindings — same singleton-only re-forward as the browse handler // above. Routed to this singleton (active node) by SiteCommunicationActor // so the dcl-manager we forward to is guaranteed to hold the live // DataConnectionActor children. Receive(msg => Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender)); // OPC UA tag-picker address-space search, secured-write execute, // and endpoint Verify — same singleton-only re-forward as the browse // handler above. SiteCommunicationActor routes these to this singleton // (active node) so the local dcl-manager is the one holding the live // DataConnectionActor children (Search/Write route there by connection name; // Verify runs a temp probe in the manager). Tell with Sender preserved is // semantically identical to Forward (ActorSelection has no Forward()). Receive(msg => Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender)); Receive(msg => Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender)); Receive(msg => Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender)); // OPC UA server-certificate trust. Trust is site-local: the // trusted-peer PKI store is per-node, so a trust/remove MUST reach BOTH // site nodes (node-a + node-b) or they diverge across failover. This // singleton fans the corresponding per-node message out to the // CertStoreActor on EVERY site node; list is answered from this node. Receive(HandleTrustServerCert); Receive(HandleRemoveServerCert); Receive(HandleListServerCerts); // UA1: cert-trust reconcile-on-join. A (re)joining site node may have // missed a trust broadcast while it was down, leaving its PKI store // divergent forever. The singleton reads its own trusted store and pushes // each cert to the joining node so the two per-node stores reconverge. Receive(HandleMemberUp); Receive(HandleSiteNodeJoined); Receive(HandleLocalCertExport); // Internal startup messages Receive(HandleStartupConfigsLoaded); // S5: re-attempt the startup deployed-config load after a transient // failure. Ignored once the load has already completed so a stale timer // (e.g. a failure retry that raced a subsequent success) cannot re-run it. Receive(_ => { if (_startupLoadCompleted) return; LoadDeployedConfigs(); }); Receive(HandleSharedScriptsLoaded); Receive(HandleStartNextBatch); // Internal enable result Receive(HandleEnableResult); // Internal deploy persistence result Receive(HandleDeployPersistenceResult); // Instance Actor init confirmation — the second half of the deploy-success // join (S6). Success is only reported once both this and persistence land. Receive(HandleInstanceActorInitialized); // Terminated signal — drains a buffered redeployment once the previous // Instance Actor has fully stopped. Receive(HandleTerminated); } /// protected override void PreStart() { base.PreStart(); _healthCollector?.SetActiveNode(true); _logger.LogInformation("DeploymentManagerActor starting — loading deployed configs from SQLite..."); // UA1: watch for site nodes coming Up so we can reconcile cert trust onto // them. Cluster-only concern — a no-op in the unit-test ActorSystem which // has no cluster provider. if (ClusterAvailable(Context.System)) { Cluster.Get(Context.System).Subscribe(Self, ClusterEvent.SubscriptionInitialStateMode.InitialStateAsEvents, typeof(ClusterEvent.MemberUp)); } LoadDeployedConfigs(); } /// /// True when this is running the cluster ref-provider. /// The unit-test harness uses the local provider, where /// would throw — the cert reconcile-on-join subscription is skipped there. /// private static bool ClusterAvailable(ActorSystem system) => system is ExtendedActorSystem ext && ext.Provider.GetType().Name.Contains("Cluster", StringComparison.Ordinal); /// /// Loads all deployed configs off the actor thread and pipes the result back as a /// . Called from and re-invoked /// by a timer when a load fails (S5) — a transient SQLite /// failure must NOT leave the site as a silent zero-instance node, so the load is retried /// at a fixed interval until it succeeds. /// private void LoadDeployedConfigs() { _configLoader().ContinueWith(t => { if (t.IsCompletedSuccessfully) return new StartupConfigsLoaded(t.Result, null); return new StartupConfigsLoaded([], t.Exception?.GetBaseException().Message); }).PipeTo(Self); } /// protected override void PostStop() { _healthCollector?.SetActiveNode(false); if (ClusterAvailable(Context.System)) { Cluster.Get(Context.System).Unsubscribe(Self); } base.PostStop(); } /// protected override SupervisorStrategy SupervisorStrategy() { return new OneForOneStrategy( maxNrOfRetries: -1, withinTimeRange: TimeSpan.FromMinutes(1), decider: Decider.From(ex => { if (ex is ActorInitializationException) { _logger.LogError(ex, "Instance Actor failed to initialize, stopping"); return Directive.Stop; } _logger.LogWarning(ex, "Instance Actor threw exception, resuming"); return Directive.Resume; })); } /// /// Processes the loaded configs from SQLite. /// /// Shared scripts must be compiled before Instance Actors are /// created, but the SQLite read and Roslyn compilation must not block the /// singleton's mailbox. The compilation is run on a background task and a /// message is piped back; only then does /// staggered Instance Actor creation begin. The deployed configs are stashed on the /// actor field in the meantime. /// private void HandleStartupConfigsLoaded(StartupConfigsLoaded msg) { if (msg.Error != null) { // S5: a transient SQLite failure must not strand the site as a silent // zero-instance node. Log and re-arm the load at a fixed interval; the // retry is unlimited and self-cancels once a load succeeds. _logger.LogError( "Failed to load deployed configs: {Error} — retrying in {Interval}", msg.Error, _startupLoadRetryInterval); Timers.StartSingleTimer( "startup-load-retry", RetryStartupLoad.Instance, _startupLoadRetryInterval); return; } // Mark the startup load complete so a stale retry timer is ignored. _startupLoadCompleted = true; var enabledConfigs = msg.Configs.Where(c => c.IsEnabled).ToList(); _deployedInstanceNames.Clear(); foreach (var c in msg.Configs) _deployedInstanceNames.Add(c.InstanceUniqueName); _logger.LogInformation( "Loaded {Total} deployed configs ({Enabled} enabled) from SQLite", msg.Configs.Count, enabledConfigs.Count); UpdateInstanceCounts(); // Load and compile shared scripts off the actor thread, then resume startup. LoadSharedScriptsFromStorage(enabledConfigs); } /// /// Once shared scripts have been compiled off-thread, begins /// staggered Instance Actor creation for the enabled configs captured at startup. /// private void HandleSharedScriptsLoaded(SharedScriptsLoaded msg) { _logger.LogInformation( "Loaded {Compiled}/{Total} shared scripts from SQLite", msg.CompiledCount, msg.TotalCount); if (msg.EnabledConfigs.Count == 0) return; // Start the first batch immediately var batchState = new BatchState(msg.EnabledConfigs, 0); Self.Tell(new StartNextBatch(batchState)); } /// /// Creates Instance Actors in batches with a configurable delay between batches /// to prevent reconnection storms on failover. /// private void HandleStartNextBatch(StartNextBatch msg) { var state = msg.State; var batchSize = _options.StartupBatchSize; var startIdx = state.NextIndex; var endIdx = Math.Min(startIdx + batchSize, state.Configs.Count); _logger.LogDebug( "Creating Instance Actors batch [{Start}..{End}) of {Total}", startIdx, endIdx, state.Configs.Count); for (var i = startIdx; i < endIdx; i++) { var config = state.Configs[i]; EnsureDclConnections(config.ConfigJson); CreateInstanceActor(config.InstanceUniqueName, config.ConfigJson); } UpdateInstanceCounts(); // Schedule next batch if there are more, using Timers (IWithTimers) if (endIdx < state.Configs.Count) { var nextState = new BatchState(state.Configs, endIdx); Timers.StartSingleTimer( "startup-batch", new StartNextBatch(nextState), TimeSpan.FromMilliseconds(_options.StartupBatchDelayMs)); } else { _logger.LogInformation( "All {Count} Instance Actors created", state.Configs.Count); } } /// /// Handles a new deployment: stores config in SQLite, clears previous static overrides, /// and creates or replaces the Instance Actor. /// /// Redeployment of an already-running instance must wait for the previous Instance /// Actor to fully terminate (including PostStop on its descendants) before the /// replacement is created — otherwise can collide on /// the still-registered child name. Instead of guessing with a fixed timer, the /// terminating child is watched and the in-flight command is buffered until the /// signal arrives. /// /// /// The actor to reply to with the eventual . /// Passed explicitly (rather than read from ) so the /// notify-and-fetch path () can supply the ORIGINAL /// central sender after the async config fetch, where Sender is no longer valid. /// The redeploy-buffer path carries it on so the buffered /// apply still replies to the right actor. /// private void HandleDeploy(DeployInstanceCommand command, IActorRef replyTo) { // Site-side compile gate (S3): a compile failure must reject the // deployment with NO partial state applied (no Instance Actor, no // persisted config) — the design spec's contract. Validation runs // synchronously on the actor thread: it is a pure prefix step of the // deploy handler, so the existing redeploy-supersede / delete-during- // redeploy ordering (which depends on strict mailbox FIFO) is preserved // exactly. It is NOT run off-thread — piping the verdict back to self // reorders concurrent deploys relative to each other and to // delete/disable commands, breaking that ordering. A deploy is an // infrequent admin command, so briefly holding the singleton for a pure // Roslyn compile is acceptable; the central deployer already Asks and // waits for the DeploymentStatusResponse. Redeploys and multi-instance // deploys of unchanged scripts hit the process-wide compile cache // (SiteScriptCompileCache), so the synchronous gate recompiles only // genuinely new code and the Instance Actor's PreStart reuses the gate's // compile (N4). var errors = _deployCompileValidator.Validate(command.FlattenedConfigurationJson); if (errors.Count > 0) { var instanceName = command.InstanceUniqueName; var joined = string.Join(" | ", errors); LogDeploymentEvent("Error", instanceName, $"Instance {instanceName} deploy rejected — site compile validation failed (deploymentId={command.DeploymentId})", joined); replyTo.Tell(new DeploymentStatusResponse( command.DeploymentId, instanceName, DeploymentStatus.Failed, "Script compilation failed on site: " + joined, DateTimeOffset.UtcNow)); return; } ProceedWithDeploy(command, replyTo); } /// /// Core deploy application (redeploy-supersede + fresh apply), reached only /// after the synchronous compile gate in passes. /// private void ProceedWithDeploy(DeployInstanceCommand command, IActorRef replyTo) { var instanceName = command.InstanceUniqueName; _logger.LogInformation( "Deploying instance {Instance}, deploymentId={DeploymentId}", instanceName, command.DeploymentId); // Redeployment replaces a running instance. Watch + stop the existing actor // and buffer this command until its Terminated signal confirms the child // (and its whole subtree) has fully stopped and freed its actor name. if (_instanceActors.TryGetValue(instanceName, out var existing)) { _instanceActors.Remove(instanceName); _pendingRedeploys[existing] = new PendingRedeploy(command, replyTo); _terminatingActorsByName[instanceName] = existing; Context.Watch(existing); Context.Stop(existing); UpdateInstanceCounts(); return; } // A deploy arriving while the previous redeploy is still // terminating (the Terminated signal hasn't fired yet) used to fall through // to ApplyDeployment(fresh), where Context.ActorOf would throw // InvalidActorNameException because the child name is still registered. // Detect the mid-termination state and overwrite the buffered pending // command (last-write-wins) so the latest deploy is applied when Terminated // arrives. The displaced sender is told Failed-superseded so it doesn't // wait forever. if (_terminatingActorsByName.TryGetValue(instanceName, out var terminatingRef)) { if (_pendingRedeploys.TryGetValue(terminatingRef, out var displaced)) { displaced.OriginalSender.Tell(new DeploymentStatusResponse( displaced.Command.DeploymentId, instanceName, DeploymentStatus.Failed, $"superseded by newer deployment {command.DeploymentId} before predecessor finished terminating", DateTimeOffset.UtcNow)); } _pendingRedeploys[terminatingRef] = new PendingRedeploy(command, replyTo); return; } // Fresh deployment — no existing actor to replace. ApplyDeployment(command, replyTo, isRedeploy: false); } /// /// Handles an Instance Actor's signal. Two cases: /// /// The child was stopped as part of a redeployment — recreate it from the /// buffered . /// The child terminated OUTSIDE the redeploy flow — an init failure or crash. /// Evict the now-dead ref from the routing map so the singleton never routes to a /// stopped actor (S6). /// /// private void HandleTerminated(Terminated terminated) { // Case 1 — redeployment drain: the predecessor of a buffered redeploy has // fully stopped and freed its actor name. if (_pendingRedeploys.Remove(terminated.ActorRef, out var pending)) { // Drop the name-keyed shadow now that the predecessor has // fully terminated and its actor name is free. Any deploy arriving after // this point sees neither _instanceActors[name] (cleared when we stopped // the predecessor) nor _terminatingActorsByName[name] (cleared here), so // ApplyDeployment + Context.ActorOf below safely reuses the name. _terminatingActorsByName.Remove(pending.Command.InstanceUniqueName); ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true); return; } // Case 2 — S6 fallback: an Instance Actor terminated unexpectedly (init // failure or crash). Expected stops (disable/delete/redeploy/persistence // rollback) all remove or replace the map entry BEFORE the child stops, so // the equality guard is false for them and this branch is inert. Only a dead // ref still mapped to this exact actor reaches here. var name = terminated.ActorRef.Path.Name; if (_instanceActors.TryGetValue(name, out var mapped) && mapped.Equals(terminated.ActorRef)) { _instanceActors.Remove(name); UpdateInstanceCounts(); _logger.LogError( "Instance Actor {Instance} terminated unexpectedly (init failure or crash) — removed from routing", name); LogDeploymentEvent("Error", name, $"Instance {name} terminated unexpectedly (init failure or crash) — removed from routing"); // S6: if the death happened while a deployment was still in flight, the // actor died during init — fail that deployment (the persistence path must // NOT later report Success) and roll back the optimistically-applied state // so the site never advertises an instance it cannot durably recover. if (_inFlightDeployments.Remove(name, out var inflight)) { inflight.ReplyTo.Tell(new DeploymentStatusResponse( inflight.DeploymentId, name, DeploymentStatus.Failed, "Instance actor failed during initialization — see site event log", DateTimeOffset.UtcNow)); if (!inflight.IsRedeploy) { _deployedInstanceNames.Remove(name); UpdateInstanceCounts(); // Remove the optimistically-persisted config row so a restart does // not re-create the failing actor. If the store already committed // (Persisted), remove it now; otherwise defer the removal until the // in-flight store completes so the remove cannot race ahead of it. if (inflight.Persisted) RollbackPersistedConfig(name); else _initFailedPendingRowRemoval.Add(name); } } } } /// /// Notify-and-fetch: handles a small central→site /// . Fetches the deployment's flattened config /// from central over HTTP via , then pipes the /// result back to self so the existing apply path runs on the actor thread with the /// ORIGINAL sender preserved (the central Ask's temp actor). The reply is the existing /// , so the central deploy completes unchanged. /// private void HandleRefreshDeployment(RefreshDeploymentCommand cmd) { // Capture the Ask temp-actor sender BEFORE the async continuation: Akka's Sender // is only valid during synchronous message handling and is no longer the original // sender once the ContinueWith/PipeTo continuation runs on a thread-pool thread. var replyTo = Sender; if (_configFetcher is null) { replyTo.Tell(new DeploymentStatusResponse( cmd.DeploymentId, cmd.InstanceUniqueName, DeploymentStatus.Failed, "Deployment config fetcher not available on this node.", DateTimeOffset.UtcNow)); return; } _logger.LogInformation( "Fetching config for deployment {DeploymentId} instance {Instance} (notify-and-fetch)", cmd.DeploymentId, cmd.InstanceUniqueName); // CancellationToken.None: the fetch is bounded by HttpClient.Timeout. On a singleton // handover mid-fetch the PipeTo lands in dead letters and the central Ask times out // (then reconciles) — acceptable, rare. _configFetcher.FetchAsync(cmd.CentralFetchBaseUrl, cmd.DeploymentId, cmd.FetchToken, CancellationToken.None) .ContinueWith(t => t.IsCompletedSuccessfully ? (object)new RefreshFetched(cmd, t.Result, replyTo) : new RefreshFetchFailed(cmd, t.Exception?.GetBaseException().Message ?? "fetch failed", replyTo)) .PipeTo(Self); } /// /// Notify-and-fetch: the config fetch succeeded — reconstruct the in-process /// apply DTO and reuse the existing apply path, /// threading the original central sender through so the /// reaches it. /// private void HandleRefreshFetched(RefreshFetched msg) { // Carry the central fetch coordinates on the apply DTO so they survive the whole // apply path (including the PendingRedeploy buffer) down to ApplyDeployment, where // they are replicated to the standby as an id-only notify-and-fetch message. var command = new DeployInstanceCommand( msg.Cmd.DeploymentId, msg.Cmd.InstanceUniqueName, msg.Cmd.RevisionHash, msg.ConfigJson, msg.Cmd.DeployedBy, msg.Cmd.Timestamp, msg.Cmd.CentralFetchBaseUrl, msg.Cmd.FetchToken); HandleDeploy(command, msg.ReplyTo); } /// /// Notify-and-fetch: the config fetch failed — report /// to the original central sender so the deploy /// completes (rather than the central Ask hanging to timeout). Nothing is applied. /// private void HandleRefreshFetchFailed(RefreshFetchFailed msg) { _logger.LogError( "Config fetch failed for deployment {DeploymentId} instance {Instance}: {Error}", msg.Cmd.DeploymentId, msg.Cmd.InstanceUniqueName, msg.Error); msg.ReplyTo.Tell(new DeploymentStatusResponse( msg.Cmd.DeploymentId, msg.Cmd.InstanceUniqueName, DeploymentStatus.Failed, $"Communication failure: {msg.Error}", DateTimeOffset.UtcNow)); } /// /// Creates the Instance Actor, persists the config, and replies to the deployer. /// A redeployment is an update of an existing instance, so the deployed-instance /// counter is only incremented for genuinely new deployments. /// /// The deployer is not told /// until SQLite persistence has committed. The site's deployed-config store is the /// durable source of truth — a config that was never persisted would be silently lost /// on the next restart/failover, so reporting Success before the row is committed is /// incorrect. The reply is sent from once /// the persistence outcome is known. /// private void ApplyDeployment(DeployInstanceCommand command, IActorRef sender, bool isRedeploy) { var instanceName = command.InstanceUniqueName; // Ensure DCL connections exist for any data-sourced attributes EnsureDclConnections(command.FlattenedConfigurationJson); // Create the Instance Actor immediately CreateInstanceActor(instanceName, command.FlattenedConfigurationJson); if (!isRedeploy) _deployedInstanceNames.Add(instanceName); UpdateInstanceCounts(); // Record the in-flight deployment so the deployer is answered exactly once: // Success only once BOTH persistence commits AND the actor confirms init, or // Failed via the S6 Terminated fallback if the Instance Actor dies during init. _inFlightDeployments[instanceName] = new InFlightDeploy { DeploymentId = command.DeploymentId, ReplyTo = sender, IsRedeploy = isRedeploy }; // Persist to SQLite and clear static overrides asynchronously Task.Run(async () => { await _storage.StoreDeployedConfigAsync( instanceName, command.FlattenedConfigurationJson, command.DeploymentId, command.RevisionHash, isEnabled: true); // Static overrides and mirrored native alarm state are reset on // redeployment per design decision — the new config may bind different // sources, and the source snapshot re-seeds the mirror on (re)subscribe. await _storage.ClearStaticOverridesAsync(instanceName); await _storage.ClearNativeAlarmsForInstanceAsync(instanceName); // Replicate to standby node — notify-and-fetch: send only the deployment id + // central fetch coordinates (NOT the config JSON). The standby fetches the // config over HTTP itself, so a large config never crosses the intra-site Akka // hop (which would silently drop on the 128 KB frame trap). When the coords are // absent (deploy paths other than RefreshDeployment), the standby fetch is a // no-op miss and reconciliation is the durable backstop. _replicationActor?.Tell(new ReplicateConfigDeploy( instanceName, command.DeploymentId, command.RevisionHash, true, command.CentralFetchBaseUrl ?? "", command.FetchToken ?? "")); return new DeployPersistenceResult( command.DeploymentId, instanceName, true, null, sender, isRedeploy); }).ContinueWith(t => { if (t.IsCompletedSuccessfully) return t.Result; return new DeployPersistenceResult( command.DeploymentId, instanceName, false, t.Exception?.GetBaseException().Message, sender, isRedeploy); }).PipeTo(Self); } /// /// Records the persistence half of the deploy-success join. On a persistence /// failure the optimistically-created Instance Actor is stopped and the counter /// rolled back and the deployer is told . On /// success the in-flight entry is marked persisted and success is reported only once /// the actor has ALSO confirmed init () — persistence /// can commit before the actor's async init has run or failed (S6). /// private void HandleDeployPersistenceResult(DeployPersistenceResult result) { // If the S6 Terminated fallback already failed this deployment (the Instance // Actor died during init), the in-flight entry is gone — swallow the now-stale // persistence result so the deployer is answered exactly once. if (!_inFlightDeployments.TryGetValue(result.InstanceName, out var inflight)) { // The fallback deferred the persisted-row rollback because the store was // still in flight then. The store has now committed (Success), so remove // the orphan row — guaranteed to run after the store, never racing it. if (_initFailedPendingRowRemoval.Remove(result.InstanceName) && result.Success) RollbackPersistedConfig(result.InstanceName); _logger.LogDebug( "Persistence result for {Instance} arrived after the deployment was already " + "resolved (init-time termination) — ignoring (deploymentId={DeploymentId}, success={Success})", result.InstanceName, result.DeploymentId, result.Success); return; } if (result.Success) { // Persistence committed — record it and complete only if init also confirmed. inflight.Persisted = true; TryCompleteDeploy(result.InstanceName, inflight); return; } // Persistence genuinely failed — fail now regardless of init state. _inFlightDeployments.Remove(result.InstanceName); _logger.LogError( "Failed to persist deployment {DeploymentId} for {Instance}: {Error}", result.DeploymentId, result.InstanceName, result.Error); // Operational `deployment` event — deploy failed. LogDeploymentEvent("Error", result.InstanceName, $"Instance {result.InstanceName} deploy failed (deploymentId={result.DeploymentId})", result.Error); // Undo the optimistic actor creation and counter bump so the site does not // advertise an instance it cannot durably recover. if (_instanceActors.Remove(result.InstanceName, out var orphan)) Context.Stop(orphan); if (!result.IsRedeploy) _deployedInstanceNames.Remove(result.InstanceName); UpdateInstanceCounts(); inflight.ReplyTo.Tell(new DeploymentStatusResponse( result.DeploymentId, result.InstanceName, DeploymentStatus.Failed, result.Error ?? "Deployment persistence failed", DateTimeOffset.UtcNow)); } /// /// Records the init half of the deploy-success join (S6): the Instance Actor /// finished PreStart and sent . Success is /// reported only once persistence has ALSO committed. Actors created outside a /// deployment (startup, enable) have no in-flight entry and are ignored here. /// private void HandleInstanceActorInitialized(InstanceActorInitialized msg) { if (_inFlightDeployments.TryGetValue(msg.InstanceUniqueName, out var inflight)) { inflight.Initialized = true; TryCompleteDeploy(msg.InstanceUniqueName, inflight); } } /// /// Completes a deployment with once BOTH the /// persistence result and the Instance Actor init confirmation have landed (S6). /// A no-op until both signals are present. /// private void TryCompleteDeploy(string instanceName, InFlightDeploy inflight) { if (!inflight.Persisted || !inflight.Initialized) return; _inFlightDeployments.Remove(instanceName); // Operational `deployment` event — deploy succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} deployed (deploymentId={inflight.DeploymentId})"); inflight.ReplyTo.Tell(new DeploymentStatusResponse( inflight.DeploymentId, instanceName, DeploymentStatus.Success, null, DateTimeOffset.UtcNow)); } /// /// Fire-and-forget removal of an optimistically-persisted deployed-config row after /// an init failure (S6). Only ever called once the row's write has committed, so it /// cannot race the store. A fault is logged, never thrown. /// private void RollbackPersistedConfig(string instanceName) { Task.Run(() => _storage.RemoveDeployedConfigAsync(instanceName)).ContinueWith(t => { if (!t.IsCompletedSuccessfully) _logger.LogError(t.Exception?.GetBaseException(), "Failed to roll back the persisted config for {Instance} after an init failure", instanceName); }); } /// /// Disables an instance: stops the actor and marks as disabled in SQLite. /// private void HandleDisable(DisableInstanceCommand command) { var instanceName = command.InstanceUniqueName; // A disable arriving mid-redeploy must cancel the buffered // redeploy. Otherwise HandleTerminated re-creates the Instance Actor and // re-stores its config with isEnabled: true when the predecessor terminates, // silently reverting the operator's disable back to enabled. Mirror the // last-write-wins handling in HandleDeploy/HandleDelete: drop the pending // command (so HandleTerminated returns early), clear the shadow, and tell the // displaced deployer it was superseded. The disable itself still persists // is_enabled = false below, which becomes the durable state. if (_terminatingActorsByName.TryGetValue(instanceName, out var terminatingRef)) { if (_pendingRedeploys.Remove(terminatingRef, out var pending)) { pending.OriginalSender.Tell(new DeploymentStatusResponse( pending.Command.DeploymentId, instanceName, DeploymentStatus.Failed, $"superseded by disable of {instanceName} before redeploy finished terminating", DateTimeOffset.UtcNow)); } _terminatingActorsByName.Remove(instanceName); } else if (_instanceActors.TryGetValue(instanceName, out var actor)) { Context.Stop(actor); _instanceActors.Remove(instanceName); } UpdateInstanceCounts(); var sender = Sender; _storage.SetInstanceEnabledAsync(instanceName, false).ContinueWith(t => { if (t.IsCompletedSuccessfully) { _replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, false)); // Operational `deployment` event — disable succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} disabled"); } else { LogDeploymentEvent("Error", instanceName, $"Instance {instanceName} disable failed", t.Exception?.GetBaseException().Message); } return new InstanceLifecycleResponse( command.CommandId, instanceName, t.IsCompletedSuccessfully, t.Exception?.GetBaseException().Message, DateTimeOffset.UtcNow); }).PipeTo(sender); _logger.LogInformation("Instance {Instance} disabled", instanceName); } /// /// Enables an instance: marks as enabled in SQLite and re-creates the Instance Actor /// from the stored config. /// private void HandleEnable(EnableInstanceCommand command) { var instanceName = command.InstanceUniqueName; var sender = Sender; Task.Run(async () => { await _storage.SetInstanceEnabledAsync(instanceName, true); _replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, true)); var configs = await _storage.GetAllDeployedConfigsAsync(); var config = configs.FirstOrDefault(c => c.InstanceUniqueName == instanceName); return new EnableResult(command, config, null, sender); }).ContinueWith(t => { if (t.IsCompletedSuccessfully) return t.Result; return new EnableResult(command, null, t.Exception?.GetBaseException().Message, sender); }).PipeTo(Self); } /// /// Processes enable result in the actor context (thread-safe). /// private void HandleEnableResult(EnableResult result) { var instanceName = result.Command.InstanceUniqueName; if (result.Error != null || result.Config == null) { var error = result.Error ?? $"No deployed config found for {instanceName}"; // Operational `deployment` event — enable failed. LogDeploymentEvent("Error", instanceName, $"Instance {instanceName} enable failed", error); result.OriginalSender.Tell(new InstanceLifecycleResponse( result.Command.CommandId, instanceName, false, error, DateTimeOffset.UtcNow)); return; } if (!_instanceActors.ContainsKey(instanceName)) { CreateInstanceActor(instanceName, result.Config.ConfigJson); } UpdateInstanceCounts(); // Operational `deployment` event — enable succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} enabled"); result.OriginalSender.Tell(new InstanceLifecycleResponse( result.Command.CommandId, instanceName, true, null, DateTimeOffset.UtcNow)); _logger.LogInformation("Instance {Instance} enabled", instanceName); } /// /// Deletes an instance: stops the actor and removes config from SQLite. /// Note: store-and-forward messages are NOT cleared per design decision. /// private void HandleDelete(DeleteInstanceCommand command) { var instanceName = command.InstanceUniqueName; // A delete arriving while a redeploy is still terminating must // be authoritative over the mid-redeploy bookkeeping. HandleDeploy already // removed the instance from _instanceActors and buffered a PendingRedeploy // keyed by the terminating ref. If we fall straight through to the // _instanceActors miss + unconditional decrement, the buffered redeploy is // left intact — so when Terminated fires, HandleTerminated calls // ApplyDeployment(isRedeploy: true) and RESURRECTS the just-deleted instance, // with the counter now inconsistent. Cancel the pending redeploy first. if (_terminatingActorsByName.TryGetValue(instanceName, out var terminatingRef)) { // Drop the buffered command so HandleTerminated's _pendingRedeploys.Remove // misses and it returns early (no resurrection). Clear the shadow too. if (_pendingRedeploys.Remove(terminatingRef, out var pending)) { pending.OriginalSender.Tell(new DeploymentStatusResponse( pending.Command.DeploymentId, instanceName, DeploymentStatus.Failed, $"superseded by delete of {instanceName} before redeploy finished terminating", DateTimeOffset.UtcNow)); } _terminatingActorsByName.Remove(instanceName); // The terminating predecessor is already being stopped by HandleDeploy; // no Context.Stop needed here. } else if (_instanceActors.TryGetValue(instanceName, out var actor)) { Context.Stop(actor); _instanceActors.Remove(instanceName); } // The deployed count is derived from the authoritative set of // deployed config names, so removing the name here decrements it. Correct for a // live, mid-redeploy, OR DISABLED instance (a disabled instance has a config row // but is absent from both in-memory maps); a delete for a never-deployed instance // removes nothing and leaves the count unchanged. Supersedes the earlier // map-presence check, which leaked the count on disabled-instance deletes. _deployedInstanceNames.Remove(instanceName); UpdateInstanceCounts(); var sender = Sender; _storage.RemoveDeployedConfigAsync(instanceName).ContinueWith(t => { if (t.IsCompletedSuccessfully) { _replicationActor?.Tell(new ReplicateConfigRemove(instanceName)); // Operational `deployment` event — delete succeeded. LogDeploymentEvent("Info", instanceName, $"Instance {instanceName} deleted"); } else { LogDeploymentEvent("Error", instanceName, $"Instance {instanceName} delete failed", t.Exception?.GetBaseException().Message); } return new InstanceLifecycleResponse( command.CommandId, instanceName, t.IsCompletedSuccessfully, t.Exception?.GetBaseException().Message, DateTimeOffset.UtcNow); }).PipeTo(sender); _logger.LogInformation("Instance {Instance} deleted", instanceName); } /// /// Fire-and-forget a deployment operational event to the optional /// on a deploy/enable/disable/delete outcome. /// Resolved optionally and never awaited so a logging failure cannot affect the /// deployment pipeline (matching the established ScriptActor/ScriptExecutionActor /// pattern). /// /// Thread-safety: the disable () and delete /// () paths call this from a /// /// continuation that runs on a thread-pool thread, NOT on the actor thread — /// so it must touch only immutable, thread-safe state. It does: the only /// field it reads is the readonly _serviceProvider captured at /// construction (the resolved is a process /// singleton). No actor-private mutable state is referenced, which is what /// makes calling it off the actor thread safe. /// /// private void LogDeploymentEvent(string severity, string instanceName, string message, string? details = null) { _ = _serviceProvider?.GetService()?.LogEventAsync( "deployment", severity, instanceName, "DeploymentManagerActor", message, details); } /// /// Answers a central query for the instance's /// currently-applied deployment identity. The site's deployed-config store /// (SQLite) is the authoritative record — it covers both enabled and /// disabled instances, and survives node restart/failover. If the instance /// has no stored config, the response reports IsDeployed = false with /// null identity so central falls through to a normal deploy. /// private void HandleDeploymentStateQuery(DeploymentStateQueryRequest request) { var sender = Sender; var instanceName = request.InstanceUniqueName; _storage.GetAllDeployedConfigsAsync().ContinueWith(t => { if (!t.IsCompletedSuccessfully) { _logger.LogError( t.Exception?.GetBaseException(), "Failed to read deployed configs for deployment state query of {Instance}", instanceName); // Treat a storage read failure as "unknown" — central falls // through to a normal deploy and relies on site-side // stale-rejection as the safety net. return new DeploymentStateQueryResponse( request.CorrelationId, instanceName, false, null, null, DateTimeOffset.UtcNow); } var config = t.Result.FirstOrDefault(c => c.InstanceUniqueName == instanceName); return config == null ? new DeploymentStateQueryResponse( request.CorrelationId, instanceName, false, null, null, DateTimeOffset.UtcNow) : new DeploymentStateQueryResponse( request.CorrelationId, instanceName, true, config.DeploymentId, config.RevisionHash, DateTimeOffset.UtcNow); }).PipeTo(sender); } // ── OPC UA server-certificate trust ── /// /// The base cluster role every site node carries (in addition to its /// per-site role site-{SiteId}). Used to enumerate the site nodes a /// trust/remove must reach so node-a and node-b PKI stores stay in lock-step /// across failover. Matches the role string set in NodeOptions.Role /// for site hosts (see AkkaHostedService.BuildRoles). /// private const string SiteClusterRole = "Site"; /// /// The per-node ask timeout for a cert broadcast. A node that does not ack /// within this window is reported as a partial failure (the command is not /// failed wholesale). /// private static readonly TimeSpan CertBroadcastTimeout = TimeSpan.FromSeconds(5); /// /// Broadcasts a trust to the per-node /// on every Up site node so both PKI stores receive the certificate. /// private void HandleTrustServerCert(TrustServerCertCommand command) => BroadcastToSiteCertStores( new WriteCertToLocalStore(command.DerBase64, command.Thumbprint), $"trust cert {command.Thumbprint} (connection {command.ConnectionName})"); /// /// Broadcasts a remove to the per-node /// on every Up site node so the certificate leaves both PKI stores. /// private void HandleRemoveServerCert(RemoveServerCertCommand command) => BroadcastToSiteCertStores( new RemoveCertFromLocalStore(command.Thumbprint), $"remove cert {command.Thumbprint}"); /// /// Lists this node's trusted + rejected PKI stores by asking the LOCAL /// (the singleton's own node). The list reflects /// the active node's view; a trust broadcast keeps the standby in sync. /// private void HandleListServerCerts(ListServerCertsCommand command) { var sender = Sender; var local = Context.ActorSelection($"/user/{CertStoreActor.WellKnownName}"); local.Ask(new ListLocalCerts(), CertBroadcastTimeout) .ContinueWith(t => { if (t.IsCompletedSuccessfully) { var ack = t.Result; return new CertTrustResult(ack.Success, ack.Error, ack.Certs); } var err = t.Exception?.GetBaseException().Message ?? "cert store did not respond"; _logger.LogWarning("Local cert store list failed: {Error}", err); return new CertTrustResult(false, err, null); }).PipeTo(sender); } /// /// Fans out to the /// on every Up site node, asks each with a short timeout, and aggregates the /// acks into one . A node that fails to ack is a /// partial failure (Success=false, first error reported) — it never throws, /// so a single unreachable standby cannot stall the singleton. /// private void BroadcastToSiteCertStores(object localMessage, string description) { var sender = Sender; var cluster = Cluster.Get(Context.System); var targets = cluster.State.Members .Where(m => m.Status == MemberStatus.Up && m.HasRole(SiteClusterRole)) .Select(m => m.Address) .ToList(); if (targets.Count == 0) { _logger.LogWarning("No Up site nodes found to {Description}; nothing trusted", description); sender.Tell(new CertTrustResult(false, "no site nodes available", null)); return; } _logger.LogInformation("Broadcasting cert op to {Count} site node(s): {Description}", targets.Count, description); var asks = targets.Select(address => { var path = new RootActorPath(address) / "user" / CertStoreActor.WellKnownName; return Context.ActorSelection(path) .Ask(localMessage, CertBroadcastTimeout) .ContinueWith(t => t.IsCompletedSuccessfully ? t.Result : new LocalCertOpAck(false, $"node {address} did not ack: {t.Exception?.GetBaseException().Message}", null)); }).ToArray(); Task.WhenAll(asks).ContinueWith(t => { // ContinueWith on WhenAll over per-task error-trapping completions // never faults, so t.Result is always populated. var acks = t.Result; var allSucceeded = acks.All(a => a.Success); var firstError = acks.FirstOrDefault(a => !a.Success)?.Error; if (!allSucceeded) { _logger.LogWarning("Cert broadcast partial/total failure ({Description}): {Error}", description, firstError); } return new CertTrustResult(allSucceeded, firstError, null); }).PipeTo(sender); } /// /// The site node most recently observed coming Up, awaiting its trusted-cert /// reconcile push. A site cluster is two nodes, so at most one peer reconciles /// at a time; a single pending slot is sufficient (a second join before the /// first export returns simply reconciles the newer node — both then converge). /// private Address? _pendingCertReconcileTarget; /// /// UA1: a site node came Up. On the singleton this is the trigger to reconcile /// cert trust onto that node. Self and non-site members are ignored — self is /// already the source of truth, and only site nodes host a CertStoreActor. /// private void HandleMemberUp(ClusterEvent.MemberUp evt) { var member = evt.Member; if (!member.HasRole(SiteClusterRole)) return; if (member.Address.Equals(Cluster.Get(Context.System).SelfAddress)) return; Self.Tell(new SiteNodeJoined(member.Address)); } /// /// Reads this node's own trusted-peer store (via the LOCAL CertStoreActor) so /// its contents can be pushed to the joined node. The export reply is handled /// by , which uses the target captured here. /// private void HandleSiteNodeJoined(SiteNodeJoined msg) { _pendingCertReconcileTarget = msg.Address; Context.ActorSelection($"/user/{CertStoreActor.WellKnownName}") .Tell(new ExportLocalTrustedCerts()); } /// /// Pushes each locally-trusted certificate to the joined node's CertStoreActor. /// Additive union only: removals are NOT reconciled here (there is no central /// persistence of cert trust yet — a logged follow-up), and each write is /// idempotent (overwrites by thumbprint), so re-pushing an already-trusted cert /// is harmless. Fire-and-forget: a failed push heals on the next MemberUp or a /// manual re-broadcast. /// private void HandleLocalCertExport(LocalCertExport export) { var target = _pendingCertReconcileTarget; _pendingCertReconcileTarget = null; if (target is null) return; if (!export.Success) { _logger.LogWarning( "Cert-trust reconcile: could not read local trusted store for joining node {Address}: {Error}", target, export.Error); return; } var basePath = new RootActorPath(target) / "user" / CertStoreActor.WellKnownName; foreach (var cert in export.Certs) { Context.ActorSelection(basePath).Tell( new WriteCertToLocalStore(Convert.ToBase64String(cert.DerBytes), cert.Thumbprint)); } _logger.LogInformation( "Cert-trust reconcile: pushed {Count} trusted cert(s) to joining node {Address}", export.Certs.Count, target); } // ── DCL connection management ── /// /// Tracks the configuration last sent to the DCL for each connection name, keyed by /// a hash of the connection's protocol/endpoints/credentials/failover count. /// A name whose hash is unchanged is skipped; a name whose config /// changed re-issues a CreateConnectionCommand so the DCL adopts the new /// configuration instead of keeping a stale connection after a redeployment. /// private readonly Dictionary _createdConnections = new(); /// /// Sets up DCL connections from the flattened config. Idempotent on unchanged /// configuration, but re-issues the create command when a connection's endpoint, /// credentials, backup endpoint, or failover retry count has changed since it was /// last sent. /// private void EnsureDclConnections(string configJson) { if (_dclManager == null) return; try { var config = System.Text.Json.JsonSerializer.Deserialize(configJson); if (config?.Connections == null) return; foreach (var (name, connConfig) in config.Connections) { EnsureDclConnection( name, connConfig.Protocol, connConfig.ConfigurationJson, connConfig.BackupConfigurationJson, connConfig.FailoverRetryCount); } } catch (Exception ex) { _logger.LogWarning(ex, "Failed to parse flattened config for DCL connections"); } } /// /// Hash-guarded DCL connection push shared by the inline /// per-instance path () and the /// system-wide artifact-deploy path (). /// Unchanged config is a no-op; a changed endpoint/credentials/backup/ /// failover-count re-issues a CreateConnectionCommand so a system- /// wide artifact-deploy makes its data-connection change live immediately /// (the artifact-deploy path previously only persisted to SQLite — the /// DCL didn't see the change until next instance redeploy or node /// restart, contradicting the "site is self-contained after artifact /// deployment" intent). /// private void EnsureDclConnection( string name, string protocol, string? primaryConfigurationJson, string? backupConfigurationJson, int failoverRetryCount) { if (_dclManager == null) return; var configHash = ComputeConnectionConfigHashCore( protocol, primaryConfigurationJson, backupConfigurationJson, failoverRetryCount); if (_createdConnections.TryGetValue(name, out var lastHash) && lastHash == configHash) return; var primaryDetails = FlattenConnectionConfig(protocol, primaryConfigurationJson); var backupDetails = string.IsNullOrEmpty(backupConfigurationJson) ? null : FlattenConnectionConfig(protocol, backupConfigurationJson); _dclManager.Tell(new Commons.Messages.DataConnection.CreateConnectionCommand( name, protocol, primaryDetails, backupDetails, failoverRetryCount)); var changed = _createdConnections.ContainsKey(name); _createdConnections[name] = configHash; _logger.LogInformation( "{Action} DCL connection {Connection} (protocol={Protocol})", changed ? "Updated" : "Created", name, protocol); } /// /// Computes a stable hash over the configuration fields that affect how the DCL /// connects, so a changed endpoint/credential/backup/failover count is detected. /// private static string ComputeConnectionConfigHash( Commons.Types.Flattening.ConnectionConfig connConfig) => ComputeConnectionConfigHashCore( connConfig.Protocol, connConfig.ConfigurationJson, connConfig.BackupConfigurationJson, connConfig.FailoverRetryCount); /// /// Field-based core so the system-wide artifact-deploy /// path (which carries protocol/config-json/backup-json/failover directly /// on ) can /// share the same hash + skip-or-resend logic as the inline-config path. /// private static string ComputeConnectionConfigHashCore( string protocol, string? primaryConfigurationJson, string? backupConfigurationJson, int failoverRetryCount) { var material = string.Join( "", protocol, primaryConfigurationJson ?? string.Empty, backupConfigurationJson ?? string.Empty, failoverRetryCount.ToString(System.Globalization.CultureInfo.InvariantCulture)); var bytes = System.Security.Cryptography.SHA256.HashData( System.Text.Encoding.UTF8.GetBytes(material)); return Convert.ToHexString(bytes); } private static IDictionary FlattenConnectionConfig(string protocol, string? json) { if (string.IsNullOrEmpty(json)) return new Dictionary(); if (string.Equals(protocol, "OpcUa", StringComparison.OrdinalIgnoreCase)) { var (config, _) = Commons.Serialization.OpcUaEndpointConfigSerializer.Deserialize(json); return Commons.Serialization.OpcUaEndpointConfigSerializer.ToFlatDict(config); } if (string.Equals(protocol, "MxGateway", StringComparison.OrdinalIgnoreCase)) { var config = Commons.Serialization.MxGatewayEndpointConfigSerializer.Deserialize(json); return Commons.Serialization.MxGatewayEndpointConfigSerializer.ToFlatDict(config); } // Fallback: assume legacy flat-dict shape for any future / unknown protocol. try { var dict = new Dictionary(); using var doc = System.Text.Json.JsonDocument.Parse(json); foreach (var prop in doc.RootElement.EnumerateObject()) dict[prop.Name] = prop.Value.ToString(); return dict; } catch { return new Dictionary(); } } // ── Shared Script Loading ── /// /// Reads and compiles all shared scripts on a background task so the /// SQLite read and Roslyn compilation never block the singleton's mailbox thread. The /// result is piped back as a message, carrying the /// enabled configs to resume staggered Instance Actor creation on the actor thread. /// private void LoadSharedScriptsFromStorage(List enabledConfigs) { Task.Run(async () => { var scripts = await _storage.GetAllSharedScriptsAsync(); var compiled = 0; foreach (var script in scripts) { if (_sharedScriptLibrary.CompileAndRegister(script.Name, script.Code)) compiled++; } return new SharedScriptsLoaded(enabledConfigs, compiled, scripts.Count); }).ContinueWith(t => { if (t.IsCompletedSuccessfully) return t.Result; _logger.LogError( t.Exception?.GetBaseException(), "Failed to load shared scripts from SQLite"); // A shared-script load failure must not abandon startup — proceed with // Instance Actor creation; scripts that need a missing shared script fail // at execution time and are recorded to the site event log. return new SharedScriptsLoaded(enabledConfigs, 0, 0); }).PipeTo(Self); } // ── Debug View routing ── private void RouteDebugViewSubscribe(SubscribeDebugViewRequest request) { if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { instanceActor.Forward(request); } else { // Set InstanceNotFound=true so the caller can distinguish // "not deployed on this site" from a deployed-but-empty instance. _logger.LogWarning( "Debug view subscribe for unknown instance {Instance}", request.InstanceUniqueName); Sender.Tell(new DebugViewSnapshot( request.InstanceUniqueName, Array.Empty(), Array.Empty(), DateTimeOffset.UtcNow, InstanceNotFound: true)); } } private void RouteDebugViewUnsubscribe(UnsubscribeDebugViewRequest request) { if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { instanceActor.Forward(request); } } private void RouteDebugSnapshot(DebugSnapshotRequest request) { if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { instanceActor.Forward(request); } else { // Set InstanceNotFound=true so the caller can distinguish // "not deployed on this site" from a deployed-but-empty instance. _logger.LogWarning( "Debug snapshot for unknown instance {Instance}", request.InstanceUniqueName); Sender.Tell(new DebugViewSnapshot( request.InstanceUniqueName, Array.Empty(), Array.Empty(), DateTimeOffset.UtcNow, InstanceNotFound: true)); } } // ── Inbound API routing ── private void RouteInboundApiCall(RouteToCallRequest request) { if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { // Convert to ScriptCallRequest and Ask the Instance Actor. // (ParentExecutionId): carry the inbound request's // ExecutionId down as ParentExecutionId so the routed script // execution can record its spawner. var scriptCall = new ScriptCallRequest( request.ScriptName, request.Parameters, 0, request.CorrelationId, ParentExecutionId: request.ParentExecutionId); var sender = Sender; instanceActor.Ask(scriptCall, TimeSpan.FromSeconds(30)) .ContinueWith(t => { if (t.IsCompletedSuccessfully) { var result = t.Result; return new RouteToCallResponse( request.CorrelationId, result.Success, // The routed script's return value crosses the Central↔Site // PROCESS boundary inside this response. A script's natural // `return new { ... }` is an anonymous type, which Akka's // cross-process serializer cannot round-trip — the reply is // silently dropped and the caller's Route.To().Call() Ask // times out even though the script succeeded. Project the // value to a plain CLR graph (Dictionary/List/primitive) that // round-trips the wire and reproduces the same JSON shape on // the inbound side (which JsonSerializer-serializes it for the // HTTP body / ReturnDefinition validation). NormalizeRoutedReturnValue(result.ReturnValue), result.ErrorMessage, DateTimeOffset.UtcNow); } return new RouteToCallResponse( request.CorrelationId, false, null, t.Exception?.GetBaseException().Message ?? "Script call timed out", DateTimeOffset.UtcNow); }).PipeTo(sender); } else { Sender.Tell(new RouteToCallResponse( request.CorrelationId, false, null, $"Instance '{request.InstanceUniqueName}' not found on this site.", DateTimeOffset.UtcNow)); } } /// /// Projects a routed script's return value to a cross-process-serializable plain /// CLR graph ( / / /// string / long / double / bool / null) via a JSON round-trip. A script's natural /// return new { ... } is a compiler-generated anonymous type that Akka's /// cross-process serializer cannot reconstruct on the receiving node, so the /// reply is silently dropped and the caller's /// Ask times out. The projected graph round-trips the wire and re-serializes to the /// same JSON shape the inbound API expects. Returns /// unchanged on a (non-expected) serialization failure so a quirky value still has a /// chance to deliver rather than being forced to null. /// private object? NormalizeRoutedReturnValue(object? value) { if (value is null) { return null; } try { var json = System.Text.Json.JsonSerializer.Serialize(value); using var doc = System.Text.Json.JsonDocument.Parse(json); return FromJsonElement(doc.RootElement); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to normalize routed script return value of type {Type} for transport; sending as-is", value.GetType().Name); return value; } } /// /// Converts a to a plain CLR value /// (string / long / double / bool / null, or nested Dictionary / List) — never a /// (which is itself not cross-process /// serializable). Companion to . /// private static object? FromJsonElement(System.Text.Json.JsonElement e) => e.ValueKind switch { System.Text.Json.JsonValueKind.Object => e.EnumerateObject().ToDictionary(p => p.Name, p => FromJsonElement(p.Value)), System.Text.Json.JsonValueKind.Array => e.EnumerateArray().Select(FromJsonElement).ToList(), System.Text.Json.JsonValueKind.String => e.GetString(), System.Text.Json.JsonValueKind.Number => e.TryGetInt64(out var l) ? l : e.GetDouble(), System.Text.Json.JsonValueKind.True => true, System.Text.Json.JsonValueKind.False => false, _ => null, }; /// /// Reads attribute values from a deployed instance for a Route.To().GetAttribute(s) /// call (or a central Test Run bound to the instance). Asks the Instance Actor /// per attribute and combines the results. /// private void RouteInboundApiGetAttributes(RouteToGetAttributesRequest request) { if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { Sender.Tell(new RouteToGetAttributesResponse( request.CorrelationId, new Dictionary(), false, $"Instance '{request.InstanceUniqueName}' not found on this site.", DateTimeOffset.UtcNow)); return; } var sender = Sender; var names = request.AttributeNames; var asks = names .Select(name => instanceActor.Ask( new GetAttributeRequest( request.CorrelationId, request.InstanceUniqueName, name, DateTimeOffset.UtcNow), TimeSpan.FromSeconds(30))) .ToArray(); Task.WhenAll(asks).ContinueWith(t => { if (t.IsCompletedSuccessfully) { var values = new Dictionary(); for (var i = 0; i < names.Count; i++) // Each attribute value crosses the Central↔Site PROCESS boundary inside // this response. For a List-typed (static or coerced) attribute the value // is a concrete generic List — a non-primitive shape Akka's cross-process // serializer cannot reliably round-trip, which would silently drop the reply // and hang the caller's Route.To().GetAttributes() Ask (same risk class as // RouteInboundApiCall / RouteInboundApiWaitForAttribute). Project each value // through the same normalizer so the wire carries a plain CLR graph // (Dictionary/List/primitive). Keys are preserved; scalars/strings/null pass // through unchanged. values[names[i]] = t.Result[i].Found ? NormalizeRoutedReturnValue(t.Result[i].Value) : null; return new RouteToGetAttributesResponse( request.CorrelationId, values, true, null, DateTimeOffset.UtcNow); } return new RouteToGetAttributesResponse( request.CorrelationId, new Dictionary(), false, t.Exception?.GetBaseException().Message ?? "Attribute read timed out", DateTimeOffset.UtcNow); }).PipeTo(sender); } /// /// Spec §6 (WD-2b): unpacks a routed /// (inbound-API Route.To().WaitForAttribute()) into the deployed /// Instance Actor's site-local and relays /// the result back. Value-equality across the wire — the predicate is null (a /// cross-process lambda cannot be routed), but RequireGoodQuality is honored /// from . The Ask is /// bounded by the wait timeout plus slack (NOT a fixed 30s), since the wait /// legitimately blocks for up to . /// private void RouteInboundApiWaitForAttribute(RouteToWaitForAttributeRequest request) { if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { Sender.Tell(new RouteToWaitForAttributeResponse( request.CorrelationId, false, null, null, false, false, $"Instance '{request.InstanceUniqueName}' not found on this site.", DateTimeOffset.UtcNow)); return; } var sender = Sender; // Routed waits are value-equality (predicate null — a cross-process lambda // cannot be routed); RequireGoodQuality is honored from the request. var inner = new WaitForAttributeRequest( request.CorrelationId, request.InstanceUniqueName, request.AttributeName, request.TargetValueEncoded, null, request.Timeout, DateTimeOffset.UtcNow, request.RequireGoodQuality); // Ask bounded by the WAIT timeout + slack — NOT a fixed 30s (the wait legitimately blocks up to request.Timeout). instanceActor.Ask(inner, request.Timeout + TimeSpan.FromSeconds(5)) .ContinueWith(t => t.IsCompletedSuccessfully // The matched value crosses the Central↔Site PROCESS boundary inside this // response. For a List-typed attribute the Instance Actor's matched // `WaitForAttributeResponse.Value` is a concrete generic List // (List/List/List/… built by TryCoerceListValue; // see InstanceActor.HandleTagValueUpdate / ResolveMatchedWaiters) — a // non-primitive shape that Akka's cross-process serializer cannot reliably // round-trip, which would silently drop the reply and hang the caller's // Ask. Project it through the same normalizer RouteInboundApiCall uses so // the wire carries a plain CLR graph (List/primitive). Scalars/strings/null // pass through unchanged. ? new RouteToWaitForAttributeResponse( request.CorrelationId, t.Result.Matched, NormalizeRoutedReturnValue(t.Result.Value), t.Result.Quality, t.Result.TimedOut, true, null, DateTimeOffset.UtcNow) : new RouteToWaitForAttributeResponse( request.CorrelationId, false, null, null, false, false, t.Exception?.GetBaseException().Message ?? "Attribute wait timed out", DateTimeOffset.UtcNow)) .PipeTo(sender); } /// /// Writes attribute values on a deployed instance for a Route.To().SetAttribute(s) /// call (or a central Test Run bound to the instance). Each write is Ask'd to the /// Instance Actor, which routes data-sourced attributes through the DCL and static /// attributes to a persisted override. The response reflects the real per-attribute /// outcome (a non-existent attribute or a failed device write reports failure), /// rather than an unconditional optimistic ack. /// private void RouteInboundApiSetAttributes(RouteToSetAttributesRequest request) { if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor)) { Sender.Tell(new RouteToSetAttributesResponse( request.CorrelationId, false, $"Instance '{request.InstanceUniqueName}' not found on this site.", DateTimeOffset.UtcNow)); return; } var sender = Sender; var correlationId = request.CorrelationId; var asks = request.AttributeValues .Select(kvp => instanceActor.Ask( new SetStaticAttributeCommand( correlationId, request.InstanceUniqueName, kvp.Key, kvp.Value, DateTimeOffset.UtcNow), TimeSpan.FromSeconds(30))) .ToArray(); Task.WhenAll(asks).ContinueWith(t => { if (!t.IsCompletedSuccessfully) return new RouteToSetAttributesResponse( correlationId, false, t.Exception?.GetBaseException().Message ?? "Attribute write timed out", DateTimeOffset.UtcNow); var failures = t.Result .Where(r => !r.Success) .Select(r => $"{r.AttributeName}: {r.ErrorMessage}") .ToArray(); return failures.Length == 0 ? new RouteToSetAttributesResponse(correlationId, true, null, DateTimeOffset.UtcNow) : new RouteToSetAttributesResponse( correlationId, false, string.Join("; ", failures), DateTimeOffset.UtcNow); }).PipeTo(sender); } /// /// Handles system-wide artifact deployment (shared scripts, external systems, etc.). /// Persists artifacts to SiteStorageService and recompiles shared scripts. /// private void HandleDeployArtifacts(DeployArtifactsCommand command) { _logger.LogInformation( "Deploying system artifacts, deploymentId={DeploymentId}", command.DeploymentId); var sender = Sender; // Capture Self before entering Task.Run: the Self/Sender/Context properties // are backed by the ambient ActorCell, which is null on a thread-pool thread, // so reading Self *inside* the lambda throws "no active ActorContext". The // data-connections branch below dispatches via this captured ref. var self = Self; Task.Run(async () => { try { // Store shared scripts and recompile if (command.SharedScripts != null) { foreach (var script in command.SharedScripts) { await _storage.StoreSharedScriptAsync(script.Name, script.Code, script.ParameterDefinitions, script.ReturnDefinition); // Shared scripts recompiled on update _sharedScriptLibrary.CompileAndRegister(script.Name, script.Code); } } // Store external system definitions if (command.ExternalSystems != null) { foreach (var es in command.ExternalSystems) { await _storage.StoreExternalSystemAsync(es.Name, es.EndpointUrl, es.AuthType, es.AuthConfiguration, es.MethodDefinitionsJson, es.TimeoutSeconds); } } // Store database connection definitions if (command.DatabaseConnections != null) { foreach (var db in command.DatabaseConnections) { await _storage.StoreDatabaseConnectionAsync(db.Name, db.ConnectionString, db.MaxRetries, db.RetryDelay); } } // Notification lists and SMTP // configuration are central-only — sites store-and-forward notifications // to central and never deliver over SMTP. Central no longer ships these // (the DeployArtifactsCommand fields stay for additive compatibility but // are always null), so the site neither persists them nor reads them. // Purge any rows a prior (pre-fix) build may have written — including the // plaintext SMTP password — so existing exposure is cleared, not just // future writes. Purge is idempotent and runs on every artifact apply. await _storage.PurgeCentralOnlyNotificationConfigAsync(); // Store data connection definitions (OPC UA endpoints, etc.) if (command.DataConnections != null) { foreach (var dc in command.DataConnections) { await _storage.StoreDataConnectionDefinitionAsync( dc.Name, dc.Protocol, dc.PrimaryConfigurationJson, dc.BackupConfigurationJson, dc.FailoverRetryCount); } // After the SQLite store, dispatch an // internal message back to the actor thread so the DCL // push runs through EnsureDclConnection — keeping the // _createdConnections hash cache mutation actor-thread- // confined while still making the change live immediately // (previously the change landed in SQLite but the DCL // kept using the stale connection until next instance // redeploy or node restart, contradicting "site is // self-contained after artifact deployment"). The // helper's hash cache skips unchanged definitions, so // the push is idempotent for re-deploys of the same // artifact bundle. self.Tell(new ApplyArtifactDataConnectionsToDcl(command.DataConnections)); } // SMTP configuration is // central-only and is never stored on a site (see the purge above). // Replicate artifacts to standby node _replicationActor?.Tell(new ReplicateArtifacts(command)); return new ArtifactDeploymentResponse( command.DeploymentId, "", true, null, DateTimeOffset.UtcNow); } catch (Exception ex) { return new ArtifactDeploymentResponse( command.DeploymentId, "", false, ex.Message, DateTimeOffset.UtcNow); } }).PipeTo(sender); } /// /// Creates a child Instance Actor with the given name and configuration JSON. /// /// The unique name of the instance to create an actor for. /// The JSON-serialized flattened configuration for the instance. internal void CreateInstanceActor(string instanceName, string configJson) { if (_instanceActors.ContainsKey(instanceName)) { _logger.LogWarning("Instance Actor {Instance} already exists, skipping creation", instanceName); return; } // Reuse the shared, host-configured logger factory // instead of allocating (and leaking) a fresh LoggerFactory per instance. var props = Props.Create(() => new InstanceActor( instanceName, configJson, _storage, _compilationService, _sharedScriptLibrary, _streamManager, _options, _loggerFactory.CreateLogger(), _dclManager, _healthCollector, _serviceProvider)); var actorRef = Context.ActorOf(props, instanceName); _instanceActors[instanceName] = actorRef; // Watch every Instance Actor so an unexpected termination (init failure or // crash) delivers a Terminated signal that evicts the dead ref from routing // (S6). Redeploy already double-watches; a double Watch is idempotent in // Akka.NET (a single Terminated is delivered). Context.Watch(actorRef); _logger.LogDebug("Created Instance Actor for {Instance}", instanceName); } /// /// Actor-thread handler that pushes artifact-deploy data /// connection definitions to the DCL via the shared /// helper. Dispatched from /// 's off-thread Task so the /// hash-cache mutation stays /// actor-thread-confined. /// private void HandleApplyArtifactDataConnectionsToDcl(ApplyArtifactDataConnectionsToDcl msg) { foreach (var dc in msg.DataConnections) { EnsureDclConnection( dc.Name, dc.Protocol, dc.PrimaryConfigurationJson, dc.BackupConfigurationJson, dc.FailoverRetryCount); } } /// /// Gets the count of active Instance Actors (for testing/diagnostics). /// internal int InstanceActorCount => _instanceActors.Count; /// /// Updates the health collector with current instance counts. /// Total deployed = _deployedInstanceNames.Count, enabled = running actors, disabled = difference. /// private void UpdateInstanceCounts() { _healthCollector?.SetInstanceCounts( deployed: _deployedInstanceNames.Count, enabled: _instanceActors.Count, disabled: _deployedInstanceNames.Count - _instanceActors.Count); } // ── Internal messages ── /// /// UA1: a site cluster node has come Up and should receive a cert-trust /// reconcile push. Derived from (self and /// non-site members already filtered out) so the reconcile flow is unit-testable /// without constructing a real cluster membership event. /// internal sealed record SiteNodeJoined(Address Address); internal record StartupConfigsLoaded(List Configs, string? Error); /// /// Internal singleton signal that re-attempts the startup deployed-config load /// after a transient SQLite failure (S5). Delivered by a single-shot Akka timer. /// internal sealed record RetryStartupLoad { /// The shared singleton instance. public static readonly RetryStartupLoad Instance = new(); private RetryStartupLoad() { } } /// /// Internal message piped back once shared scripts have been compiled off-thread /// Carries the enabled configs so staggered Instance Actor /// creation resumes on the actor thread. /// internal record SharedScriptsLoaded( List EnabledConfigs, int CompiledCount, int TotalCount); internal record StartNextBatch(BatchState State); internal record BatchState(List Configs, int NextIndex); internal record EnableResult( EnableInstanceCommand Command, DeployedInstance? Config, string? Error, IActorRef OriginalSender); internal record DeployPersistenceResult( string DeploymentId, string InstanceName, bool Success, string? Error, IActorRef OriginalSender, bool IsRedeploy); /// /// A redeployment command buffered until the previous Instance Actor terminates. /// internal record PendingRedeploy(DeployInstanceCommand Command, IActorRef OriginalSender); /// /// Join state for a deployment awaiting confirmation. Deployment Success is /// reported only once BOTH signals land: (the config committed /// to SQLite) AND (the Instance Actor finished PreStart and /// sent ). Persistence can commit before /// the actor's asynchronous init has run or failed, so persistence alone is not enough /// to declare success (S6). An actor that dies during init never sets /// ; the S6 Terminated fallback fails that deployment instead. /// gates rollback so a failed init rolls back the /// deployed-instance set and the persisted row only for a genuinely new deployment. /// internal sealed class InFlightDeploy { /// The deployment id echoed back in the status response. public required string DeploymentId { get; init; } /// The deployer to answer exactly once. public required IActorRef ReplyTo { get; init; } /// Whether this is a redeploy (an update) rather than a new deployment. public required bool IsRedeploy { get; init; } /// Set when the config has committed to SQLite. public bool Persisted { get; set; } /// Set when the Instance Actor has finished PreStart successfully. public bool Initialized { get; set; } } /// /// Notify-and-fetch: piped back to self when the deployment's flattened /// config has been fetched from central. Carries the original central sender so the /// reused apply path replies to it. /// private sealed record RefreshFetched(RefreshDeploymentCommand Cmd, string ConfigJson, IActorRef ReplyTo); /// /// Notify-and-fetch: piped back to self when the config fetch failed. /// Carries the original central sender so a Failed status is reported to it. /// private sealed record RefreshFetchFailed(RefreshDeploymentCommand Cmd, string Error, IActorRef ReplyTo); /// /// Internal message dispatched from /// 's off-thread persistence task back /// onto the actor thread, so the DCL push (and its hash-cache mutation) /// runs through without crossing /// thread-confinement boundaries. /// internal record ApplyArtifactDataConnectionsToDcl( IReadOnlyList DataConnections); }