dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
52cd4fb0966e0552e4e5df524d25e91f2743412f
ScadaBridge/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs
T

3803 lines
194 KiB
C#
Raw Blame History

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.Repositories;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Secrets;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
/// <summary>
/// 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.
/// </summary>
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;
/// <summary>
/// WP2.6a: the shared cache backing <c>SiteExternalSystemRepository</c> reads.
/// Invalidated wholesale after <see cref="HandleDeployArtifacts"/> applies
/// external-system changes so a script's next call re-reads fresh definitions/method
/// lists instead of serving a stale cached snapshot. Null in tests that do not wire
/// external-system caching — the invalidation call is then simply skipped.
/// </summary>
private readonly ExternalSystemDefinitionCache? _externalSystemCache;
/// <summary>
/// Task 10 (site-secrets artifact deploy): per-node mirror reconciled synchronously by
/// <see cref="HandleDeployArtifacts"/> right after the shipped secrets are written to
/// <c>deployed_secrets</c>, so the reply's <c>SecretsApplied</c> reflects this node's own
/// apply/verify outcome rather than the next periodic pass. Optional/null on nodes/tests
/// that never register secret deployment — a shipped <c>command.Secrets</c> with no mirror
/// registered fails the deploy (see <see cref="HandleDeployArtifacts"/>).
/// </summary>
private readonly DeployedSecretsMirror? _secretsMirror;
private readonly SiteStreamManager? _streamManager;
private readonly SiteRuntimeOptions _options;
private readonly ILogger<DeploymentManagerActor> _logger;
/// <summary>
/// Shared logger factory used to mint <see cref="InstanceActor"/> loggers.
/// Reused across every <see cref="CreateInstanceActor"/>
/// 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.
/// </summary>
private readonly ILoggerFactory _loggerFactory;
private readonly IActorRef? _dclManager;
private readonly ISiteHealthCollector? _healthCollector;
private readonly IServiceProvider? _serviceProvider;
/// <summary>
/// Notify-and-fetch: fetches a deployment's flattened config from central
/// over HTTP when a <see cref="RefreshDeploymentCommand"/> arrives. Optional — null on
/// nodes/tests that never receive a refresh; the active site path supplies it via DI.
/// </summary>
private readonly IDeploymentConfigFetcher? _configFetcher;
/// <summary>
/// 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.
/// </summary>
private readonly TimeSpan _startupLoadRetryInterval;
/// <summary>
/// Loads all deployed configs at startup. Defaults to
/// <see cref="SiteStorageService.GetAllDeployedConfigsAsync"/>; overridable via the
/// ctor so a test can drive a transient load failure (S5).
/// </summary>
private readonly Func<Task<List<DeployedInstance>>> _configLoader;
/// <summary>
/// Reads a single deployed config row by instance name. Defaults to
/// <see cref="SiteStorageService.GetDeployedConfigAsync"/>; overridable via the ctor so a
/// test can hold the read open mid-flight (row 101's ensure-started race tests).
/// </summary>
/// <remarks>
/// Register P1-4: the row-67 ensure-started heal used to read through
/// <see cref="_configLoader"/> — the same full-table-including-ConfigJson load the boot
/// sequence uses — just to <c>FirstOrDefault</c> one instance. A volume-restore reconcile
/// that raises hundreds of <see cref="EnsureInstanceStartedCommand"/>s in a row turned that
/// into hundreds of full-table reads. This seam is deliberately separate from
/// <see cref="_configLoader"/>: the two are read at different cardinalities (all rows for
/// boot, one row per heal) and a test overriding one must not accidentally gate the other.
/// </remarks>
private readonly Func<string, Task<DeployedInstance?>> _deployedConfigLoader;
/// <summary>
/// Reads the durable <c>data_connection_definitions</c> rows (deferred register rows 89 + 90).
/// Defaults to <see cref="SiteStorageService.GetAllDataConnectionDefinitionsAsync"/>;
/// overridable via the ctor so a test can drive a read failure.
/// <para>
/// This table had NO production reader before rows 89/90 — it was written by the artifact
/// deploy and never read back, which is precisely why a definition change survived only as
/// long as the singleton incarnation that applied it.
/// </para>
/// </summary>
private readonly Func<Task<List<StoredDataConnectionDefinition>>> _connectionDefinitionLoader;
/// <summary>
/// Set once the startup deployed-config load has succeeded. Guards against a
/// stale <see cref="RetryStartupLoad"/> re-running the load after success.
/// </summary>
private bool _startupLoadCompleted;
/// <summary>
/// How many consecutive boot-time store reads may be re-dispatched for being superseded
/// mid-flight (row 101) before one is applied anyway. Purely a livelock backstop — see
/// <see cref="HandleStartupConfigsLoaded"/>.
/// </summary>
private const int MaxConsecutiveStaleStartupLoads = 5;
/// <summary>
/// Consecutive boot-time store reads discarded as superseded. Reset the moment a fresh one
/// lands, so the cap only ever counts an unbroken run.
/// </summary>
private int _consecutiveStaleStartupLoads;
private readonly Dictionary<string, IActorRef> _instanceActors = new();
/// <summary>
/// Tracks Instance Actors that are terminating as part of a redeployment, keyed by
/// the terminating actor ref. The buffered command is applied once <see cref="Terminated"/>
/// confirms the child has fully stopped and its final persistence acknowledgement arrives.
/// </summary>
private readonly Dictionary<IActorRef, PendingRedeploy> _pendingRedeploys = new();
private readonly Dictionary<IActorRef, InstanceActor.StaticOverridesFlushed> _redeployFlushes = new();
private readonly HashSet<IActorRef> _redeployTerminated = new();
private readonly HashSet<IActorRef> _redeployStopsInFlight = new();
private readonly Queue<IActorRef> _redeployStopQueue = new();
private sealed record LifecycleStop(string Name, string CommandId, bool Delete, IActorRef ReplyTo);
private sealed record LifecycleStopPersisted(IActorRef Actor, Exception? Error);
private readonly Dictionary<IActorRef, LifecycleStop> _lifecycleStops = new();
private readonly HashSet<string> _lifecycleStopNames = new();
/// <summary>
/// Name → terminating actor ref shadow of <see cref="_pendingRedeploys"/>.
/// Required because a third <see cref="DeployInstanceCommand"/> for the same instance
/// arriving WHILE a redeploy is still mid-termination would otherwise see
/// <c>_instanceActors.TryGetValue == false</c> and fall through to
/// <see cref="ApplyDeployment"/> + <see cref="CreateInstanceActor"/>, where
/// <c>Context.ActorOf(props, instanceName)</c> throws <c>InvalidActorNameException</c>
/// — the child name is still registered until the <see cref="Terminated"/> 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
/// <see cref="HandleDeploy"/> detect the mid-termination state and overwrite the
/// buffered pending command (last-write-wins) instead of trying to create a fresh actor.
/// Cleared in <see cref="HandleTerminated"/> alongside <see cref="_pendingRedeploys"/>.
/// </summary>
private readonly Dictionary<string, IActorRef> _terminatingActorsByName = new();
/// <summary>
/// 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 <see cref="_instanceActors"/> and <see cref="_terminatingActorsByName"/>.
/// </summary>
private readonly HashSet<string> _deployedInstanceNames = new();
/// <summary>
/// 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:
/// <list type="bullet">
/// <item><see cref="HandleDeployPersistenceResult"/> on a normal persistence outcome,
/// which then replies Success/Failed;</item>
/// <item>the <see cref="HandleTerminated"/> S6 fallback when the actor dies during
/// init, which replies Failed and rolls back the optimistic state.</item>
/// </list>
/// Removing the entry in exactly one of those paths guarantees the deployer receives
/// exactly one <see cref="DeploymentStatusResponse"/> regardless of ordering.
/// </summary>
private readonly Dictionary<string, InFlightDeploy> _inFlightDeployments = new();
/// <summary>
/// Instances whose deployment was failed by the S6 Terminated fallback WHILE their
/// optimistic <c>StoreDeployedConfigAsync</c> 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
/// <see cref="HandleDeployPersistenceResult"/> when the store's result finally lands.
/// </summary>
private readonly HashSet<string> _initFailedPendingRowRemoval = new();
/// <summary>
/// WP3.1 warm-then-gate: instances whose deploy compile is being warmed off the actor
/// thread, keyed by instance name. Presence is the per-instance in-flight guard that keeps
/// same-instance command ordering intact while OTHER instances' commands flow freely.
/// Entries are added in <see cref="HandleDeploy"/> and removed in
/// <see cref="HandleDeployCompileWarmed"/> — one or the other always runs, because the
/// warm task pipes its result back unconditionally (it swallows its own exceptions).
/// </summary>
private readonly Dictionary<string, DeployWarmState> _deployWarms = new();
/// <summary>
/// WP3.1: script-execution scheduler seam (#18) — the pool this actor grows as instances
/// deploy. Null selects the process-wide shared scheduler; tests inject their own so
/// sizing assertions do not perturb (or depend on) the process-wide singleton.
/// </summary>
private readonly ScriptExecutionScheduler? _scriptScheduler;
/// <summary>
/// Deferred register row 72: the config revision hash each RUNNING Instance Actor was
/// created from, keyed by instance name. Written by <see cref="CreateInstanceActor"/> —
/// the single creation funnel — and compared against the stored row by the instance
/// reconcile sweep to detect an actor still serving a config the durable store has since
/// replaced.
/// <para>
/// Reads are always guarded by <c>_instanceActors.ContainsKey</c>, so a leftover entry for
/// an instance that has stopped is never consulted, and a stop-then-recreate overwrites it.
/// An instance with no entry (only reachable if a future creation path bypasses the funnel)
/// is treated as "revision unknown" and is deliberately never restarted.
/// </para>
/// </summary>
private readonly Dictionary<string, string> _runningRevisions = new();
/// <summary>
/// The stored revision hash the instance reconcile sweep has already ACTED on for an
/// instance, keyed by instance name — <b>one repair attempt per (instance, revision)
/// pair</b>, for both the row-78 start and the row-72 restart.
/// <para>
/// This is what keeps a periodic self-heal from becoming a restart loop. An instance whose
/// Instance Actor dies during init leaves its config row in place, so without this the
/// sweep would recreate it — and re-announce it — every tick forever; likewise a stored
/// config the site compile gate rejects would be re-attempted forever. The divergence is
/// still surfaced either way: the attempt itself emits a Warning site event, and a failed
/// restart emits a second one. A genuinely NEW revision (the normal case: central redeploys
/// something better) is a different key and is always attempted.
/// </para>
/// </summary>
private readonly Dictionary<string, string> _sweepAttemptedRevisions = new();
/// <summary>
/// Interval of the instance reconcile sweep (rows 78 + 72);
/// <see cref="TimeSpan.Zero"/> disables it. See
/// <see cref="SiteRuntimeOptions.InstanceSweepIntervalSeconds"/>.
/// </summary>
private readonly TimeSpan _instanceSweepInterval;
/// <summary>
/// True once boot-time Instance Actor creation has finished (every staggered batch created,
/// or there were no enabled configs to create). The sweep is inert until then: a tick landing
/// between two staggered batches would otherwise "heal" instances the boot load is already
/// about to create, emitting a Warning site event for a node that is merely still starting.
/// </summary>
private bool _startupInstanceCreationComplete;
/// <summary>
/// True while a sweep's off-thread storage read is outstanding. A slow read must not let
/// ticks stack up into overlapping reads (and duplicate repairs) on a large site.
/// </summary>
private bool _instanceSweepInFlight;
/// <summary>Akka timer scheduler injected by the framework via <see cref="IWithTimers"/>.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Initializes the actor, wires all message handlers, and prepares for staggered Instance Actor startup.</summary>
/// <param name="storage">Site SQLite storage service for deployed configuration persistence.</param>
/// <param name="compilationService">Script compilation service used when creating Instance Actors.</param>
/// <param name="sharedScriptLibrary">Shared script library compiled before Instance Actors are created.</param>
/// <param name="streamManager">Optional site-wide stream manager; null in unit tests.</param>
/// <param name="options">Site runtime options (startup batch size, stagger interval, etc.).</param>
/// <param name="logger">Logger instance.</param>
/// <param name="dclManager">Optional Data Connection Layer manager actor.</param>
/// <param name="replicationActor">Optional standby replication actor.</param>
/// <param name="healthCollector">Optional site health collector.</param>
/// <param name="serviceProvider">Optional service provider for resolving per-instance services.</param>
/// <param name="loggerFactory">Optional logger factory for creating Instance Actor loggers.</param>
/// <param name="configFetcher">
/// Optional notify-and-fetch config fetcher. Required for the
/// <see cref="RefreshDeploymentCommand"/> path; null on nodes/tests that never
/// receive a refresh.
/// </param>
/// <param name="startupLoadRetryInterval">Optional retry interval used when loading
/// deployed configuration at startup; defaults to 5 seconds when null.</param>
/// <param name="configLoader">Optional override for loading all deployed configurations
/// at startup; defaults to reading from <paramref name="storage"/>. Primarily for tests.</param>
/// <param name="externalSystemCache">
/// WP2.6a: the shared <see cref="ExternalSystemDefinitionCache"/> backing
/// <c>SiteExternalSystemRepository</c> reads, invalidated after this actor applies
/// external-system changes. Optional/null in tests that do not exercise external-system
/// caching.
/// </param>
/// <param name="scriptScheduler">
/// WP3.1: optional script-execution scheduler override (#18). This actor grows the pool
/// towards <see cref="ScriptExecutionScheduler.ComputeTargetThreads"/> on every instance-
/// count change; null uses the process-wide shared scheduler.
/// </param>
/// <param name="connectionDefinitionLoader">
/// Rows 89 + 90: optional override for reading the durable
/// <c>data_connection_definitions</c> rows that arbitrate the live DCL configuration;
/// defaults to reading from <paramref name="storage"/>. Primarily for tests that need to
/// drive a read failure.
/// </param>
/// <param name="deployedConfigLoader">
/// Register P1-4: optional override for the single-row deployed-config read the row-67
/// ensure-started heal uses; defaults to reading from <paramref name="storage"/>. Primarily
/// for tests that need to hold the read open mid-flight to drive the row-101 race.
/// </param>
/// <param name="secretsMirror">
/// Task 10 (site-secrets artifact deploy): this node's <see cref="DeployedSecretsMirror"/>,
/// reconciled synchronously by <see cref="HandleDeployArtifacts"/> right after a shipped
/// <c>command.Secrets</c> set is written to <c>deployed_secrets</c>. Optional/null on
/// nodes/tests that never register secret deployment.
/// </param>
public DeploymentManagerActor(
SiteStorageService storage,
ScriptCompilationService compilationService,
SharedScriptLibrary sharedScriptLibrary,
SiteStreamManager? streamManager,
SiteRuntimeOptions options,
ILogger<DeploymentManagerActor> logger,
IActorRef? dclManager = null,
ISiteHealthCollector? healthCollector = null,
IServiceProvider? serviceProvider = null,
ILoggerFactory? loggerFactory = null,
IDeploymentConfigFetcher? configFetcher = null,
TimeSpan? startupLoadRetryInterval = null,
Func<Task<List<DeployedInstance>>>? configLoader = null,
ExternalSystemDefinitionCache? externalSystemCache = null,
ScriptExecutionScheduler? scriptScheduler = null,
Func<Task<List<StoredDataConnectionDefinition>>>? connectionDefinitionLoader = null,
Func<string, Task<DeployedInstance?>>? deployedConfigLoader = null,
DeployedSecretsMirror? secretsMirror = null)
{
_scriptScheduler = scriptScheduler;
_secretsMirror = secretsMirror;
_storage = storage;
_compilationService = compilationService;
_deployCompileValidator = new DeployCompileValidator(compilationService);
_sharedScriptLibrary = sharedScriptLibrary;
_externalSystemCache = externalSystemCache;
_streamManager = streamManager;
_options = options;
_dclManager = dclManager;
_healthCollector = healthCollector;
_serviceProvider = serviceProvider;
_configFetcher = configFetcher;
_startupLoadRetryInterval = startupLoadRetryInterval ?? TimeSpan.FromSeconds(5);
_instanceSweepInterval = options.InstanceSweepIntervalSeconds > 0
? TimeSpan.FromSeconds(options.InstanceSweepIntervalSeconds)
: TimeSpan.Zero;
_configLoader = configLoader ?? (() => _storage.GetAllDeployedConfigsAsync());
_deployedConfigLoader = deployedConfigLoader ?? (name => _storage.GetDeployedConfigAsync(name));
_connectionDefinitionLoader = connectionDefinitionLoader
?? (() => _storage.GetAllDataConnectionDefinitionsAsync());
_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<DeployInstanceCommand>(cmd => HandleDeploy(cmd, Sender));
Receive<DisableInstanceCommand>(cmd => HandleDisable(cmd, Sender));
Receive<EnableInstanceCommand>(cmd => HandleEnable(cmd, Sender));
Receive<DeleteInstanceCommand>(cmd => HandleDelete(cmd, Sender));
// WP3.1 warm-then-gate: the off-thread compile warm for a deploy has finished, so
// the (now all-cache-hit) synchronous compile gate can run on the actor thread.
Receive<DeployCompileWarmed>(HandleDeployCompileWarmed);
// WP3.1: a staggered startup batch's compiles have been pre-warmed off-thread;
// create the batch's Instance Actors now that their PreStart compiles are hits.
Receive<BatchCompileWarmed>(msg => CreateInstanceActorBatch(msg.Batch));
// 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<RefreshDeploymentCommand>(HandleRefreshDeployment);
Receive<RefreshFetched>(HandleRefreshFetched);
Receive<RefreshFetchFailed>(HandleRefreshFetchFailed);
// Reconcile self-heal: a config the node was missing at boot has just been
// fetched + stored by SiteReconciliationActor, which heals the FILE only. Start the
// Instance Actor too, so a missing-config boot no longer needs a SECOND restart to
// run. Idempotent and enabled-state-aware; the durable row is the authority.
Receive<EnsureInstanceStartedCommand>(HandleEnsureInstanceStarted);
Receive<EnsureInstanceStartedResult>(HandleEnsureInstanceStartedResult);
// Instance reconcile sweep (deferred register rows 78 + 72). Periodically re-reads the
// durable rows and repairs the two ways the runtime can silently diverge from them:
// an enabled config with no running actor (row 78), and a running actor whose revision
// no longer matches the stored one (row 72).
Receive<InstanceSweepTick>(_ => HandleInstanceSweepTick());
Receive<InstanceSweepLoaded>(HandleInstanceSweepLoaded);
// Terminal reply of a sweep-driven stale-revision restart. Every OTHER deploy path
// replies to central (or to a test probe); only the sweep makes this actor its own
// deployer, so a response landing here is always its own and is surfaced, never
// dead-lettered.
Receive<DeploymentStatusResponse>(HandleSweepRestartResult);
// 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<DeploymentStateQueryRequest>(HandleDeploymentStateQuery);
// Handle system-wide artifact deployment
Receive<DeployArtifactsCommand>(HandleDeployArtifacts);
// Artifact-deploy DCL push, dispatched back from the
// off-thread persistence task so the hash-cache mutation stays
// actor-thread-confined.
Receive<ApplyArtifactDataConnectionsToDcl>(HandleApplyArtifactDataConnectionsToDcl);
// Debug View — route to Instance Actors
Receive<SubscribeDebugViewRequest>(RouteDebugViewSubscribe);
Receive<UnsubscribeDebugViewRequest>(RouteDebugViewUnsubscribe);
Receive<DebugSnapshotRequest>(RouteDebugSnapshot);
// Inbound API Route.To().Call() — route to Instance Actors
Receive<RouteToCallRequest>(RouteInboundApiCall);
Receive<RouteToGetAttributesRequest>(RouteInboundApiGetAttributes);
Receive<RouteToSetAttributesRequest>(RouteInboundApiSetAttributes);
Receive<RouteToWaitForAttributeRequest>(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<BrowseNodeCommand>(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<ReadTagValuesCommand>(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<SearchAddressSpaceCommand>(msg =>
Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender));
Receive<Commons.Messages.DataConnection.WriteTagRequest>(msg =>
Context.ActorSelection("/user/dcl-manager").Tell(msg, Sender));
Receive<VerifyEndpointCommand>(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<TrustServerCertCommand>(HandleTrustServerCert);
Receive<RemoveServerCertCommand>(HandleRemoveServerCert);
Receive<ListServerCertsCommand>(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<ClusterEvent.MemberUp>(HandleMemberUp);
Receive<SiteNodeJoined>(HandleSiteNodeJoined);
Receive<LocalCertExport>(HandleLocalCertExport);
// Internal startup messages
Receive<StartupConfigsLoaded>(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<RetryStartupLoad>(_ =>
{
if (_startupLoadCompleted)
return;
LoadDeployedConfigs();
});
Receive<SharedScriptsLoaded>(HandleSharedScriptsLoaded);
Receive<StartNextBatch>(HandleStartNextBatch);
// Internal enable result
Receive<EnableResult>(HandleEnableResult);
// Internal deploy persistence result
Receive<DeployPersistenceResult>(HandleDeployPersistenceResult);
// Row 101: an off-thread mutation of a snapshot table has committed. The only work is
// the actor-thread generation bump — see _storeWriteGeneration.
Receive<StoreWriteCompleted>(msg => BumpStoreWriteGeneration(msg.Source));
// Instance Actor init confirmation — the second half of the deploy-success
// join (S6). Success is only reported once both this and persistence land.
Receive<InstanceActorInitialized>(HandleInstanceActorInitialized);
// Terminated signal — drains a buffered redeployment once the previous
// Instance Actor has fully stopped.
Receive<Terminated>(HandleTerminated);
Receive<LifecycleStopPersisted>(message =>
{
if (!_lifecycleStops.Remove(message.Actor, out var stop)) return;
_lifecycleStopNames.Remove(stop.Name);
_terminatingActorsByName.Remove(stop.Name);
BumpStoreWriteGeneration("lifecycle stop committed");
if (stop.Delete && message.Error is null)
{
_deployedInstanceNames.Remove(stop.Name);
_runningRevisions.Remove(stop.Name);
_sweepAttemptedRevisions.Remove(stop.Name);
}
UpdateInstanceCounts();
var action = stop.Delete ? "deleted" : "disabled";
LogDeploymentEvent(message.Error is null ? "Info" : "Error", stop.Name,
$"Instance {stop.Name} {action}", message.Error?.GetBaseException().Message);
stop.ReplyTo.Tell(new InstanceLifecycleResponse(stop.CommandId, stop.Name,
message.Error is null, message.Error?.GetBaseException().Message, DateTimeOffset.UtcNow));
});
Receive<InstanceActor.StaticOverridesFlushed>(message =>
{
if (!_pendingRedeploys.ContainsKey(message.Actor) && !_redeployStopsInFlight.Contains(message.Actor) && !_lifecycleStops.ContainsKey(message.Actor)) return;
_redeployFlushes[message.Actor] = message;
TryFinishRedeployStop(message.Actor);
});
}
/// <inheritdoc />
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));
}
// Instance reconcile sweep (rows 78 + 72). Armed here rather than after the startup
// load so it survives a load that never succeeds; the handler itself is inert until
// boot-time Instance Actor creation has finished.
if (_instanceSweepInterval > TimeSpan.Zero)
{
Timers.StartPeriodicTimer(
"instance-sweep", InstanceSweepTick.Instance,
_instanceSweepInterval, _instanceSweepInterval);
}
LoadDeployedConfigs();
}
/// <summary>
/// True when this <see cref="ActorSystem"/> is running the cluster ref-provider.
/// The unit-test harness uses the local provider, where <see cref="Cluster.Get"/>
/// would throw — the cert reconcile-on-join subscription is skipped there.
/// </summary>
private static bool ClusterAvailable(ActorSystem system) =>
system is ExtendedActorSystem ext &&
ext.Provider.GetType().Name.Contains("Cluster", StringComparison.Ordinal);
/// <summary>
/// Loads all deployed configs off the actor thread and pipes the result back as a
/// <see cref="StartupConfigsLoaded"/>. Called from <see cref="PreStart"/> and re-invoked
/// by a <see cref="RetryStartupLoad"/> 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.
/// </summary>
private void LoadDeployedConfigs()
{
// Row 101: capture on the actor thread BEFORE the off-thread read starts, so the landing
// handler can tell whether this actor mutated either snapshot table while it was in
// flight.
var generationAtRead = _storeWriteGeneration;
ReadStoreSnapshotAsync().ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
return new StartupConfigsLoaded(
t.Result.Configs, null, t.Result.Definitions, generationAtRead);
return new StartupConfigsLoaded([], t.Exception?.GetBaseException().Message);
}).PipeTo(Self);
}
/// <summary>
/// Reads the two durable tables the Deployment Manager arbitrates over — the deployed
/// instance configs and the <c>data_connection_definitions</c> rows — in ONE off-thread
/// task, so the boot load and every sweep tick see a single consistent snapshot and the
/// actor thread is never blocked on SQLite.
///
/// <para>
/// <b>The two reads fail differently on purpose.</b> A failed CONFIG read propagates: it is
/// the existing S5 path and re-arms the whole load at a fixed interval. A failed DEFINITIONS
/// read is swallowed into a <see langword="null"/> result, because the DCL arbitration overlay
/// must be left EXACTLY as it was rather than cleared — an overlay cleared by a transient
/// read failure would hand arbitration back to the flattened snapshots for one pass and then
/// take it away again on the next successful read, which is the recreate-on-every-flip
/// ping-pong the whole design exists to make impossible.
/// </para>
/// </summary>
private async Task<(List<DeployedInstance> Configs, IReadOnlyList<StoredDataConnectionDefinition>? Definitions)>
ReadStoreSnapshotAsync()
{
var configs = await _configLoader().ConfigureAwait(false);
IReadOnlyList<StoredDataConnectionDefinition>? definitions = null;
try
{
definitions = await _connectionDefinitionLoader().ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(
ex, "Could not read the data connection definitions — the DCL arbitration "
+ "overlay keeps its previous contents until the next read succeeds");
}
return (configs, definitions);
}
/// <inheritdoc />
protected override void PostStop()
{
_healthCollector?.SetActiveNode(false);
if (ClusterAvailable(Context.System))
{
Cluster.Get(Context.System).Unsubscribe(Self);
}
base.PostStop();
}
/// <inheritdoc />
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;
}));
}
/// <summary>
/// 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
/// <see cref="SharedScriptsLoaded"/> message is piped back; only then does
/// staggered Instance Actor creation begin. The deployed configs are stashed on the
/// actor field in the meantime.
/// </summary>
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;
}
// Row 101, boot half: a snapshot invalidated while it was in flight cannot simply be
// DROPPED the way the sweep's is — the boot load has to complete or the node stays a
// silent zero-instance node — so it is re-read instead. The re-dispatch captures the
// current generation, so it can only be stale again if yet another mutation lands in
// ITS window; that requires a fresh deploy/enable/disable/delete per attempt, which is
// externally driven and finite in practice. The counter is the belt-and-braces cap
// against a pathological arrival rate: after MaxConsecutiveStaleStartupLoads attempts
// the snapshot is applied anyway with a Warning, because a boot that never finishes is
// strictly worse than one that starts from a slightly superseded view — the instance
// reconcile sweep re-reads the store every tick and converges it afterwards.
if (IsStoreSnapshotStale(msg.StoreGenerationAtRead, "boot load"))
{
// A completed boot load means this is a late duplicate (the S5 retry timer racing a
// success), not the boot the node is waiting on. Re-reading for it would loop
// pointlessly; the sweep owns convergence from here.
if (_startupLoadCompleted)
return;
if (++_consecutiveStaleStartupLoads <= MaxConsecutiveStaleStartupLoads)
{
LoadDeployedConfigs();
return;
}
_logger.LogWarning(
"The boot-time deployed-config read has come back superseded {Attempts} times in "
+ "a row — applying it anyway so startup completes; the instance reconcile sweep "
+ "converges any difference on its next tick",
_consecutiveStaleStartupLoads);
}
_consecutiveStaleStartupLoads = 0;
// Mark the startup load complete so a stale retry timer is ignored.
_startupLoadCompleted = true;
// Rows 89 + 90: seed the DCL arbitration overlay BEFORE any Instance Actor is created,
// so the boot batch's EnsureDclConnections already resolves to the definition-owned
// config. This is the failover half of the row-89 fix: a singleton incarnation starting
// on the peer node rebuilds the whole DCL from flattened snapshots, and without the
// overlay in place first it would rebuild them STALE and only correct itself a sweep
// tick later — a needless create-then-diverge on every failover.
ApplyConnectionDefinitionOverlay(msg.ConnectionDefinitions);
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);
}
/// <summary>
/// Once shared scripts have been compiled off-thread, begins
/// staggered Instance Actor creation for the enabled configs captured at startup.
/// </summary>
private void HandleSharedScriptsLoaded(SharedScriptsLoaded msg)
{
_logger.LogInformation(
"Loaded {Compiled}/{Total} shared scripts from SQLite",
msg.CompiledCount, msg.TotalCount);
if (msg.EnabledConfigs.Count == 0)
{
// Nothing to create, so boot-time creation is trivially complete and the instance
// reconcile sweep may run. This is exactly the row-78 live case: the node booted
// with an empty deployed_configurations table and CDC replication seeds it moments
// later, with no boot load left to notice.
_startupInstanceCreationComplete = true;
return;
}
// Start the first batch immediately
var batchState = new BatchState(msg.EnabledConfigs, 0);
Self.Tell(new StartNextBatch(batchState));
}
/// <summary>
/// WP3.1: pre-warms one staggered startup batch's script and trigger-expression compiles
/// off the actor thread, then pipes <see cref="BatchCompileWarmed"/> so
/// <see cref="CreateInstanceActorBatch"/> can create the batch's Instance Actors with
/// every <c>PreStart</c> compile already a cache hit.
///
/// <para>This closes the long-standing "per-instance compilation during staggered startup"
/// gap: on failover, each Instance Actor used to Roslyn-compile its own scripts inside
/// <c>PreStart</c>, serialising a site's whole recovery behind compilation. The cost is one
/// extra message per batch.</para>
/// </summary>
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);
var validator = _deployCompileValidator;
var batchConfigs = new string[endIdx - startIdx];
for (var i = startIdx; i < endIdx; i++)
batchConfigs[i - startIdx] = state.Configs[i].ConfigJson;
Task.Run(() =>
{
foreach (var configJson in batchConfigs)
{
try
{
// Verdict discarded: startup does not gate on compile failures (a failing
// script is logged by the Instance Actor and leaves the rest running).
// This call exists only to populate the compile cache owned by the
// shared ScriptCompilationService (SiteScriptCompileCache).
validator.Validate(configJson);
}
catch
{
// Best-effort warm — a throw here just means that instance's PreStart
// compiles the hard way, exactly as it did before WP3.1.
}
}
return new BatchCompileWarmed(msg);
}).PipeTo(Self);
}
/// <summary>
/// Creates one batch's Instance Actors (after its compiles have been pre-warmed) and
/// schedules the next batch, with a configurable delay between batches to prevent
/// reconnection storms on failover.
/// </summary>
private void CreateInstanceActorBatch(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];
// CreateInstanceActor ensures this config's DCL connections (register row 85).
CreateInstanceActor(config.InstanceUniqueName, config.ConfigJson, config.RevisionHash);
}
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);
// Boot-time creation is done — the instance reconcile sweep may now run without
// racing the staggered batches it would otherwise mistake for a divergence.
_startupInstanceCreationComplete = true;
}
}
/// <summary>
/// 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 <see cref="Context.ActorOf"/> 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
/// <see cref="Terminated"/> signal arrives.
/// </summary>
/// <param name="replyTo">
/// The actor to reply to with the eventual <see cref="DeploymentStatusResponse"/>.
/// Passed explicitly (rather than read from <see cref="ActorBase.Sender"/>) so the
/// notify-and-fetch path (<see cref="HandleRefreshFetched"/>) can supply the ORIGINAL
/// central sender after the async config fetch, where <c>Sender</c> is no longer valid.
/// The redeploy-buffer path carries it on <see cref="PendingRedeploy"/> so the buffered
/// apply still replies to the right actor.
/// </param>
private void HandleDeploy(DeployInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
if (_lifecycleStopNames.Contains(instanceName))
{
replyTo.Tell(new DeploymentStatusResponse(command.DeploymentId, instanceName, DeploymentStatus.Failed,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return;
}
// WP3.1 warm-then-gate. The S3 gate below must stay synchronous on the actor thread
// (mailbox FIFO is what makes redeploy-supersede and delete-during-redeploy correct),
// but the Roslyn compile it performs used to hold the whole singleton — every OTHER
// instance's commands stalled behind one instance's compile. So the compile is WARMED
// off-thread first and the synchronous gate then re-runs as pure cache hits.
//
// Ordering is preserved by an explicit per-instance in-flight guard: while a warm is
// in flight for instance X, further mutating commands for X are queued here rather
// than racing ahead. Commands for OTHER instances flow freely, which is the whole
// point — and safe, because cross-instance ordering was never guaranteed to callers.
if (_deployWarms.TryGetValue(instanceName, out var warming))
{
// Nothing queued behind the pending deploy yet: plain last-write-wins, and the
// displaced deployer is told it was superseded so it never waits out its Ask —
// the same contract as the mid-termination redeploy buffer below.
if (warming.Buffered.Count == 0)
{
warming.ReplyTo.Tell(new DeploymentStatusResponse(
warming.Command.DeploymentId, instanceName, DeploymentStatus.Failed,
$"superseded by newer deployment {command.DeploymentId} before the site compile gate ran",
DateTimeOffset.UtcNow));
warming.Command = command;
warming.ReplyTo = replyTo;
// Warm the NEW config; the in-flight warm's result is recognised as stale by
// the reference check in HandleDeployCompileWarmed and dropped.
StartDeployCompileWarm(command);
return;
}
// A delete/disable/enable is already queued behind the pending deploy —
// superseding now would reorder it past this deploy, so queue instead.
warming.Buffered.Add(new BufferedInstanceCommand(command, replyTo));
return;
}
_deployWarms[instanceName] = new DeployWarmState(command, replyTo);
StartDeployCompileWarm(command);
}
/// <summary>
/// Compiles the deployment's scripts off the actor thread purely to populate the compile
/// cache owned by the shared <c>ScriptCompilationService</c> singleton, then pipes
/// <see cref="DeployCompileWarmed"/> back so the authoritative gate can run on the actor
/// thread against a warm cache. The warm is best-effort: its verdict is discarded and any
/// exception swallowed, because <see cref="RunDeployGateAndProceed"/> re-derives the
/// verdict (and reproduces any failure) exactly as it did before WP3.1.
/// </summary>
private void StartDeployCompileWarm(DeployInstanceCommand command)
{
var validator = _deployCompileValidator;
var configJson = command.FlattenedConfigurationJson;
Task.Run(() =>
{
try
{
validator.Validate(configJson);
}
catch (Exception ex)
{
_logger.LogDebug(ex,
"Deploy compile warm for {Instance} threw; the synchronous gate will re-run it",
command.InstanceUniqueName);
}
return new DeployCompileWarmed(command);
}).PipeTo(Self);
}
/// <summary>
/// Runs the authoritative compile gate for a warmed deployment on the actor thread, then
/// drains any commands queued for that instance during the warm — inline and in arrival
/// order, NOT by re-telling <c>Self</c>, which would put them behind messages that landed
/// in the mailbox during the warm and so reorder same-instance commands.
/// </summary>
private void HandleDeployCompileWarmed(DeployCompileWarmed msg)
{
var instanceName = msg.Command.InstanceUniqueName;
if (!_deployWarms.TryGetValue(instanceName, out var warming))
return;
// A newer deploy superseded this one while its warm was running; that supersede
// started its own warm, so this result is stale and must not apply a dead command.
if (!ReferenceEquals(warming.Command, msg.Command))
return;
_deployWarms.Remove(instanceName);
RunDeployGateAndProceed(warming.Command, warming.ReplyTo);
foreach (var queued in warming.Buffered)
DispatchBufferedCommand(queued);
}
/// <summary>
/// Queues a mutating lifecycle command for an instance whose deploy is mid-compile-warm.
/// Returns <see langword="true"/> when the command was queued (the caller must return
/// immediately), <see langword="false"/> when no warm is in flight and the caller should
/// proceed normally.
/// </summary>
private bool TryBufferDuringDeployWarm(string instanceName, object command, IActorRef replyTo)
{
if (!_deployWarms.TryGetValue(instanceName, out var warming)) return false;
warming.Buffered.Add(new BufferedInstanceCommand(command, replyTo));
return true;
}
/// <summary>
/// Re-dispatches one command queued during a deploy compile warm, restoring the original
/// sender. A queued deploy legitimately opens a NEW warm window, in which case the
/// commands after it re-queue behind that one — ordering still holds.
/// </summary>
private void DispatchBufferedCommand(BufferedInstanceCommand queued)
{
switch (queued.Command)
{
case DeployInstanceCommand deploy:
HandleDeploy(deploy, queued.Sender);
break;
case DeleteInstanceCommand delete:
HandleDelete(delete, queued.Sender);
break;
case DisableInstanceCommand disable:
HandleDisable(disable, queued.Sender);
break;
case EnableInstanceCommand enable:
HandleEnable(enable, queued.Sender);
break;
default:
_logger.LogWarning(
"Unhandled buffered command type {Type} for a deploy compile warm — dropped",
queued.Command.GetType().Name);
break;
}
}
/// <summary>
/// The authoritative site-side compile gate (S3) plus the deploy application. Reached
/// only from <see cref="HandleDeployCompileWarmed"/>, i.e. after the same compile has
/// been warmed off-thread.
/// </summary>
private void RunDeployGateAndProceed(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 still runs
// synchronously on the actor thread as a pure prefix step of the deploy
// application, so the existing redeploy-supersede / delete-during-redeploy
// ordering (which depends on strict mailbox FIFO) is preserved exactly.
// WP3.1 removed the COST rather than the ordering: the same compile has
// already been warmed off-thread (StartDeployCompileWarm), so this call is
// pure cache hits and no longer holds the singleton — while the per-instance
// warm guard keeps this instance's own command ordering intact. Redeploys and
// multi-instance deploys of unchanged scripts hit the compile cache
// (SiteScriptCompileCache, owned by the ScriptCompilationService singleton this
// actor shares with every Instance Actor) too, 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);
}
/// <summary>
/// Core deploy application (redeploy-supersede + fresh apply), reached only
/// after the synchronous compile gate in <see cref="HandleDeploy"/> passes.
/// </summary>
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);
_redeployStopQueue.Enqueue(existing);
StartQueuedRedeployStops();
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);
}
private bool DeferLifecycleStop(string name, string commandId, bool delete, IActorRef replyTo)
{
if (_lifecycleStopNames.Contains(name))
{
replyTo.Tell(new InstanceLifecycleResponse(commandId, name, false,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return true;
}
if (!_terminatingActorsByName.TryGetValue(name, out var actor) && !_instanceActors.TryGetValue(name, out actor))
return false;
if (_pendingRedeploys.Remove(actor, out var pending))
pending.OriginalSender.Tell(new DeploymentStatusResponse(pending.Command.DeploymentId, name,
DeploymentStatus.Failed, "Superseded by instance lifecycle command.", DateTimeOffset.UtcNow));
_instanceActors.Remove(name);
_terminatingActorsByName[name] = actor;
_lifecycleStopNames.Add(name);
_lifecycleStops[actor] = new LifecycleStop(name, commandId, delete, replyTo);
BumpStoreWriteGeneration("lifecycle stop dispatch");
Context.Watch(actor);
Context.Stop(actor);
UpdateInstanceCounts();
// Termination and flush may both already have arrived for a canceled
// redeploy. Reconcile immediately as well as on either completion signal.
TryFinishRedeployStop(actor);
return true;
}
private void StartQueuedRedeployStops()
{
while (_redeployStopsInFlight.Count < 4 && _redeployStopQueue.TryDequeue(out var actor))
{
if (!_pendingRedeploys.ContainsKey(actor)) continue;
_redeployStopsInFlight.Add(actor);
Context.Stop(actor);
}
}
private void TryFinishRedeployStop(IActorRef actor)
{
if (!_redeployTerminated.Contains(actor) || !_redeployFlushes.Remove(actor, out var flush)) return;
_redeployTerminated.Remove(actor);
_redeployStopsInFlight.Remove(actor);
if (_lifecycleStops.TryGetValue(actor, out var lifecycle))
{
var persistence = Task.Run(async () =>
{
if (flush.Error is not null) throw new InvalidOperationException("Final static override persistence failed.", flush.Error);
if (lifecycle.Delete) await _storage.RemoveDeployedConfigAsync(lifecycle.Name).ConfigureAwait(false);
else await _storage.SetInstanceEnabledAsync(lifecycle.Name, false).ConfigureAwait(false);
});
persistence.PipeTo(Self, Self, () => new LifecycleStopPersisted(actor, null),
error => new LifecycleStopPersisted(actor, error));
StartQueuedRedeployStops();
return;
}
if (_pendingRedeploys.Remove(actor, out var pending))
{
_terminatingActorsByName.Remove(pending.Command.InstanceUniqueName);
if (flush.Error is null)
ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true);
else
pending.OriginalSender.Tell(new DeploymentStatusResponse(pending.Command.DeploymentId,
pending.Command.InstanceUniqueName, DeploymentStatus.Failed,
"Predecessor static override flush failed; redeployment was not applied.", DateTimeOffset.UtcNow));
}
StartQueuedRedeployStops();
}
/// <summary>
/// Handles an Instance Actor's <see cref="Terminated"/> signal. Two cases:
/// <list type="number">
/// <item>The child was stopped as part of a redeployment — recreate it from the
/// buffered <see cref="DeployInstanceCommand"/>.</item>
/// <item>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).</item>
/// </list>
/// </summary>
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.ContainsKey(terminated.ActorRef) || _redeployStopsInFlight.Contains(terminated.ActorRef) || _lifecycleStops.ContainsKey(terminated.ActorRef))
{
_redeployTerminated.Add(terminated.ActorRef);
TryFinishRedeployStop(terminated.ActorRef);
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);
}
}
}
}
/// <summary>
/// Notify-and-fetch: handles a small central→site
/// <see cref="RefreshDeploymentCommand"/>. Fetches the deployment's flattened config
/// from central over HTTP via <see cref="IDeploymentConfigFetcher"/>, 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
/// <see cref="DeploymentStatusResponse"/>, so the central deploy completes unchanged.
/// </summary>
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);
}
/// <summary>
/// Notify-and-fetch: the config fetch succeeded — reconstruct the in-process
/// <see cref="DeployInstanceCommand"/> apply DTO and reuse the existing apply path,
/// threading the original central sender through so the
/// <see cref="DeploymentStatusResponse"/> reaches it.
/// </summary>
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);
}
/// <summary>
/// Notify-and-fetch: the config fetch failed — report
/// <see cref="DeploymentStatus.Failed"/> to the original central sender so the deploy
/// completes (rather than the central Ask hanging to timeout). Nothing is applied.
/// </summary>
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));
}
/// <summary>
/// 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 <b>not</b> told <see cref="DeploymentStatus.Success"/>
/// 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 <see cref="HandleDeployPersistenceResult"/> once
/// the persistence outcome is known.
/// </summary>
private void ApplyDeployment(DeployInstanceCommand command, IActorRef sender, bool isRedeploy)
{
var instanceName = command.InstanceUniqueName;
// 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. Dispatched HERE, BEFORE the successor exists,
// and handed to it as a gate (arch-review 2026-08-19 P1-2). It used to run inside the
// persistence task below, i.e. after the actor was already created AND queued behind a
// full config-JSON write, so the successor's PreStart override load routinely won the
// race and applied the PREDECESSOR's overrides — including keys absent from the new
// config — to in-memory state for the actor's whole lifetime, while the disk ended up
// cleared. That combination is worse than either half alone: the running instance
// silently disagreed with its own store, and a later restart flipped the values.
var staticStateResetGate = Task.Run(async () =>
{
await _storage.ClearStaticOverridesAsync(instanceName);
await _storage.ClearNativeAlarmsForInstanceAsync(instanceName);
});
// Create the Instance Actor immediately. CreateInstanceActor first ensures the DCL
// connections this config's data-sourced attributes subscribe through (register row 85).
CreateInstanceActor(
instanceName, command.FlattenedConfigurationJson, command.RevisionHash,
staticStateResetGate);
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
};
// Row 101 (initiation half): a deployed_configurations write is about to start off the
// actor thread, so every store read currently in flight predates it.
BumpStoreWriteGeneration("deploy persist dispatch");
// Persist to SQLite and clear static overrides asynchronously
Task.Run(async () =>
{
await _storage.StoreDeployedConfigAsync(
instanceName,
command.FlattenedConfigurationJson,
command.DeploymentId,
command.RevisionHash,
isEnabled: true);
// Join the reset dispatched ahead of actor creation, so a failed clear still
// fails the deployment exactly as it did when the clear ran inline here.
await staticStateResetGate;
// No explicit replication step: deployed_configurations is a replicated table,
// so the write above is captured and shipped to the peer like any other row.
// This used to be a notify-and-fetch Tell — the standby was sent the deployment
// id plus central fetch coordinates and pulled the config over HTTP itself,
// because a large config would silently drop on the intra-site Akka hop's
// 128 KB frame trap. Replication carries the row directly and has no such limit,
// so the standby no longer fetches on deploy. (SiteReconciliationActor still
// fetches at node startup when central reports gaps — a different path.)
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);
}
/// <summary>
/// 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 <see cref="DeploymentStatus.Failed"/>. On
/// success the in-flight entry is marked persisted and success is reported only once
/// the actor has ALSO confirmed init (<see cref="TryCompleteDeploy"/>) — persistence
/// can commit before the actor's async init has run or failed (S6).
/// </summary>
private void HandleDeployPersistenceResult(DeployPersistenceResult result)
{
// Row 101 (completion half): this message is created only after the store write has
// finished, so bumping here is the point that provably happens-after the commit. Bumped
// unconditionally — a FAILED write may still have committed the config row before a
// later step threw, so "failed" is not "nothing changed".
BumpStoreWriteGeneration("deploy persist 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));
}
/// <summary>
/// Records the init half of the deploy-success join (S6): the Instance Actor
/// finished PreStart and sent <see cref="InstanceActorInitialized"/>. 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.
/// </summary>
private void HandleInstanceActorInitialized(InstanceActorInitialized msg)
{
if (_inFlightDeployments.TryGetValue(msg.InstanceUniqueName, out var inflight))
{
inflight.Initialized = true;
TryCompleteDeploy(msg.InstanceUniqueName, inflight);
}
}
/// <summary>
/// Completes a deployment with <see cref="DeploymentStatus.Success"/> once BOTH the
/// persistence result and the Instance Actor init confirmation have landed (S6).
/// A no-op until both signals are present.
/// </summary>
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));
}
/// <summary>
/// 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.
/// </summary>
private void RollbackPersistedConfig(string instanceName)
{
// Row 101 (initiation half) — the row is about to disappear from under any in-flight read.
BumpStoreWriteGeneration("init-failure rollback dispatch");
var removal = Task.Run(() => _storage.RemoveDeployedConfigAsync(instanceName));
removal.ContinueWith(t =>
{
if (!t.IsCompletedSuccessfully)
_logger.LogError(t.Exception?.GetBaseException(),
"Failed to roll back the persisted config for {Instance} after an init failure",
instanceName);
});
// Row 101 (completion half): this path has no other actor-thread completion message, so
// one exists purely to bump the generation after the DELETE has committed. Without it a
// sweep read dispatched between the two points above would still see the rolled-back row
// and start an instance whose deployment had just failed.
removal.ContinueWith(_ => new StoreWriteCompleted("init-failure rollback")).PipeTo(Self);
}
/// <summary>
/// Disables an instance: stops the actor and marks as disabled in SQLite.
/// </summary>
private void HandleDisable(DisableInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
// WP3.1 warm-then-gate: a deploy for this instance is mid-compile-warm; queue this
// command so same-instance ordering is preserved (see TryBufferDuringDeployWarm).
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
if (DeferLifecycleStop(instanceName, command.CommandId, false, replyTo)) return;
UpdateInstanceCounts();
var sender = replyTo;
// Row 101 (initiation half): the actor has just stopped the Instance Actor while the
// durable row still reads enabled, which is exactly the shape the row-78 sweep repairs.
// Any store read in flight would therefore "prove" a divergence that does not exist and
// restart the instance the operator just disabled.
BumpStoreWriteGeneration("disable dispatch");
var disableWrite = _storage.SetInstanceEnabledAsync(instanceName, false);
// Row 101 (completion half) — see StoreWriteCompleted. The is_enabled UPDATE has landed
// by the time this message is created, closing the window between the bump above and the
// commit.
disableWrite.ContinueWith(_ => new StoreWriteCompleted("disable")).PipeTo(Self);
disableWrite.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
// 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);
}
/// <summary>
/// Enables an instance: marks as enabled in SQLite and re-creates the Instance Actor
/// from the stored config.
/// </summary>
private void HandleEnable(EnableInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
if (_lifecycleStopNames.Contains(instanceName))
{
replyTo.Tell(new InstanceLifecycleResponse(command.CommandId, instanceName, false,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return;
}
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
var sender = replyTo;
// Row 101 (initiation half) — an is_enabled UPDATE is about to start off-thread.
BumpStoreWriteGeneration("enable dispatch");
Task.Run(async () =>
{
await _storage.SetInstanceEnabledAsync(instanceName, true);
// Row P1-4: a single-row read — this used to materialize every deployed config
// (including every ConfigJson blob) just to FirstOrDefault this one instance.
var config = await _storage.GetDeployedConfigAsync(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);
}
/// <summary>
/// Processes enable result in the actor context (thread-safe).
/// </summary>
private void HandleEnableResult(EnableResult result)
{
// Row 101 (completion half): the is_enabled UPDATE has committed by the time this
// message exists. Unconditional for the same reason as the deploy result.
BumpStoreWriteGeneration("enable 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, result.Config.RevisionHash);
}
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);
}
/// <summary>
/// Reconcile self-heal (deferred register row 67): makes sure an instance whose config was
/// (re)stored out-of-band is actually RUNNING.
///
/// <para>
/// <b>Why this exists.</b> Instance Actors were created from exactly two paths — the
/// boot-time SQLite load (<see cref="HandleStartupConfigsLoaded"/>) and an explicit
/// deploy/enable command. <c>SiteReconciliationActor</c>'s node-STARTUP fetch runs AFTER
/// that boot load has already read storage, so a config it healed into the file started no
/// actor: the instance stayed dead until the next restart's load finally picked it up.
/// This handler closes that gap without touching the deploy pipeline.
/// </para>
///
/// <para>
/// <b>Guards.</b> The command is advisory and fire-and-forget, so every reason to do
/// nothing is a silent return: the actor is already running (idempotent — both nodes'
/// reconcile actors legitimately raise this for the same instance), or an in-flight
/// deploy/redeploy owns the instance and will create the actor itself. Deferring to the
/// deploy paths is what keeps the deploy gate's ordering guarantees and the
/// all-or-nothing-per-instance rule intact — this never applies a config, it only starts
/// one that is already fully persisted.
/// </para>
/// </summary>
private void HandleEnsureInstanceStarted(EnsureInstanceStartedCommand command)
{
var instanceName = command.InstanceUniqueName;
if (!ShouldAttemptEnsureStart(instanceName))
return;
// Re-read the durable row off the actor thread. The stored row — NOT the requesting
// message — is the authority on the config body AND on enabled/disabled state, so a
// disable that landed before this read still wins and the instance stays stopped.
//
// Row 101: a disable/delete that lands WHILE the read is in flight is a different
// problem, and the row itself cannot catch it — the read already saw the enabled row,
// and neither Disable nor Delete leaves any of the three markers
// ShouldAttemptEnsureStart re-checks. So capture the generation on the actor thread
// BEFORE the read starts and let the landing handler discard a superseded snapshot,
// exactly as the boot load and the sweep do. Without it the heal could start an
// instance the operator had just disabled or deleted, PERMANENTLY: the sweep only ever
// walks store rows, so it has no runtime→store stop leg to undo it.
var generationAtRead = _storeWriteGeneration;
Task.Run(async () =>
{
// Register P1-4: read a single row through _deployedConfigLoader (defaults to
// _storage.GetDeployedConfigAsync — a WHERE instance_unique_name = @name read, not
// the full-table-including-ConfigJson load the boot sequence uses), which still lets
// a test hold the read open and land the result behind a committed disable/delete.
var config = await _deployedConfigLoader(instanceName);
return new EnsureInstanceStartedResult(command, config, null, generationAtRead);
}).ContinueWith(t => t.IsCompletedSuccessfully
? t.Result
: new EnsureInstanceStartedResult(command, null, t.Exception?.GetBaseException().Message))
.PipeTo(Self);
}
/// <summary>
/// True when a deploy/redeploy/delete pipeline currently OWNS the instance — a compile warm
/// is in flight, a predecessor is still terminating, or a deployment is awaiting its
/// persistence/init join. Every advisory self-heal (row 67's ensure-started and the row
/// 78/72 sweep) defers to those paths: they are authoritative, they will create or replace
/// the actor themselves, and stepping in front of them would break the deploy gate's
/// ordering guarantees and the all-or-nothing-per-instance rule.
/// </summary>
private bool InstanceIsOwnedByADeployPipeline(string instanceName) =>
_deployWarms.ContainsKey(instanceName) ||
_terminatingActorsByName.ContainsKey(instanceName) ||
_inFlightDeployments.ContainsKey(instanceName);
/// <summary>
/// Shared guard for the two halves of the ensure-started round trip. Re-evaluated after
/// the storage read because the instance's state can change while that read is in flight
/// (a deploy or delete may have arrived on the mailbox in between).
/// </summary>
private bool ShouldAttemptEnsureStart(string instanceName) =>
!_instanceActors.ContainsKey(instanceName) &&
!InstanceIsOwnedByADeployPipeline(instanceName);
/// <summary>
/// Actor-thread half of the reconcile self-heal: creates the Instance Actor for a config
/// that is present and enabled in storage but has no running actor, and SURFACES the heal
/// (log + operational site event) so "config stored, actor not running" is no longer
/// silent.
/// </summary>
private void HandleEnsureInstanceStartedResult(EnsureInstanceStartedResult result)
{
var instanceName = result.Command.InstanceUniqueName;
var reason = result.Command.Reason;
if (result.Error != null)
{
_logger.LogWarning(
"Ensure-started ({Reason}) for {Instance} could not read the deployed config: {Error}",
reason, instanceName, result.Error);
return;
}
// Row 101, ensure-started half: this node disabled, deleted or redeployed an instance
// while the read was in flight, so the snapshot predates state the actor has already
// advanced past. Discard it — starting from it would resurrect an instance the operator
// just stopped, and nothing would ever stop it again. Nothing is lost by dropping it:
// the heal is advisory, and the instance reconcile sweep re-reads the store every tick.
if (IsStoreSnapshotStale(result.StoreGenerationAtRead, "ensure-started"))
return;
// No stored row: nothing to start. Normal when the config was superseded/removed
// between the reconcile fetch and this read, and on an UNREPLICATED site whose peer
// node healed its own file — this singleton's storage legitimately has no such row.
if (result.Config == null)
{
_logger.LogDebug(
"Ensure-started ({Reason}) for {Instance}: no deployed config on this node — nothing to start",
reason, instanceName);
return;
}
// Disabled instances stay stopped. The heal restores the config row either way; a
// disabled instance having a row but no actor is the correct steady state.
if (!result.Config.IsEnabled)
{
_logger.LogDebug(
"Ensure-started ({Reason}) for {Instance}: instance is disabled — leaving it stopped",
reason, instanceName);
return;
}
// Re-check the guards: a deploy/delete for this instance may have arrived while the
// storage read was in flight, and it is authoritative over this advisory start.
if (!ShouldAttemptEnsureStart(instanceName))
{
_logger.LogDebug(
"Ensure-started ({Reason}) for {Instance}: already running or superseded by an in-flight deploy",
reason, instanceName);
return;
}
StartStoredInstance(result.Config, reason,
"Config was missing at node boot and was restored by reconciliation; "
+ "the Instance Actor was not running and has now been started.");
}
/// <summary>
/// The one place an advisory self-heal turns a stored, enabled config into a running
/// Instance Actor — shared by the row-67 ensure-started handler and the row-78 sweep so
/// both produce identical bookkeeping and identical operator-facing surfacing. Callers
/// are responsible for the guards; this method assumes they have passed.
/// </summary>
/// <param name="config">The durable row to start from — the authority on config and revision.</param>
/// <param name="reason">Provenance, carried verbatim into the log line and the site event.</param>
/// <param name="details">Longer explanation attached to the site event.</param>
private void StartStoredInstance(DeployedInstance config, string reason, string details)
{
var instanceName = config.InstanceUniqueName;
CreateInstanceActor(instanceName, config.ConfigJson, config.RevisionHash);
_deployedInstanceNames.Add(instanceName);
UpdateInstanceCounts();
// Surface the heal. Warning (not Info) on purpose: reaching here means an enabled,
// successfully-deployed instance was NOT running while its config sat in the store —
// the condition is repaired, but it is abnormal and worth an operator's attention.
_logger.LogWarning(
"Instance {Instance} started by {Reason} self-heal — its config is present in the "
+ "deployed-config store but no Instance Actor was running until now",
instanceName, reason);
LogDeploymentEvent("Warning", instanceName,
$"Instance {instanceName} started by {reason} self-heal", details);
}
/// <summary>Provenance string carried on every repair the instance reconcile sweep makes.</summary>
private const string SweepReason = "deployment-manager sweep";
/// <summary>
/// Instance reconcile sweep, tick half (deferred register rows 78 + 72). Kicks off an
/// off-thread read of the durable <c>deployed_configurations</c> rows; the comparison and
/// every repair happen on the actor thread in
/// <see cref="HandleInstanceSweepLoaded"/>.
///
/// <para>
/// <b>Why a periodic sweep and not a LocalDb change hook.</b> Row 78's live symptom was a
/// config that arrived by CDC <i>replication</i>: the sync applier writes
/// <c>deployed_configurations</c> rows straight into the file, which is invisible to the
/// boot-time load (already run) and to the startup reconcile (central reports no gap,
/// because central's view is correct — the site really does have the config). A hook on the
/// replication apply would cover exactly that one arrival path, and only for the half of
/// the fleet that runs with replication enabled. It is also not available: <c>ILocalDb</c>
/// exposes no change-notification or applied-rows callback at all, so a hook would mean a
/// new upstream library capability plus an in-process bus from the sync applier into this
/// singleton. A sweep over the durable rows is the arrival-path-agnostic form of the same
/// check — replication seeding, a post-boot snapshot resync, an operator restoring a file,
/// and any future path all converge on "the store says X, the runtime says Y" — and it
/// inherits row 67's guards unchanged.
/// </para>
///
/// <para>
/// <b>Inert until boot-time creation finishes.</b> A tick landing between two staggered
/// startup batches would see instances the boot load is about to create and "heal" them,
/// announcing a fault on a node that is merely still starting.
/// </para>
/// </summary>
private void HandleInstanceSweepTick()
{
if (!_startupInstanceCreationComplete || _instanceSweepInFlight)
return;
_instanceSweepInFlight = true;
// Row 101: capture on the actor thread BEFORE the off-thread read starts.
var generationAtRead = _storeWriteGeneration;
ReadStoreSnapshotAsync().ContinueWith(t => t.IsCompletedSuccessfully
? new InstanceSweepLoaded(t.Result.Configs, null, t.Result.Definitions, generationAtRead)
: new InstanceSweepLoaded([], t.Exception?.GetBaseException().Message))
.PipeTo(Self);
}
/// <summary>
/// Instance reconcile sweep, actor-thread half: compares the durable rows against the
/// running Instance Actors and repairs the two divergences the register recorded.
///
/// <para>
/// A sweep that finds nothing is <b>completely silent</b> — no log line above Trace, no site
/// event — because that is the case on every tick of a healthy site.
/// </para>
/// </summary>
private void HandleInstanceSweepLoaded(InstanceSweepLoaded msg)
{
_instanceSweepInFlight = false;
if (msg.Error != null)
{
_logger.LogWarning(
"Instance reconcile sweep could not read the deployed configs: {Error}", msg.Error);
return;
}
// Row 101, sweep half: this node mutated one of the two tables while the read was in
// flight, so the snapshot predates state the actor has already advanced past. Discard it
// WHOLE — no overlay apply, no row-78 start, no row-72 restart. Every repair this sweep
// makes is derived from "the store says X, the runtime says Y", and a snapshot older
// than the runtime inverts that comparison: it would restart an instance BACKWARD onto
// the revision a deploy just replaced, or resurrect one a disable/delete just stopped.
// Nothing is lost by waiting — the next tick's read starts after the write.
if (IsStoreSnapshotStale(msg.StoreGenerationAtRead, "instance sweep"))
return;
// Rows 89 + 90: refresh the DCL arbitration overlay FIRST, so any instance this same
// sweep pass goes on to start resolves its connections against the current definitions
// rather than against the snapshot frozen into its flattened config. This is the
// convergence path for a definition row that arrived by LocalDb CDC replication: no new
// cross-node message exists (or is needed) — whichever node runs the singleton reads the
// row out of its own file on the next tick.
ApplyConnectionDefinitionOverlay(msg.ConnectionDefinitions);
foreach (var config in msg.Configs)
{
var instanceName = config.InstanceUniqueName;
// A disabled instance legitimately has a config row and no running actor; that is
// its correct steady state, not a divergence.
if (!config.IsEnabled)
continue;
// Defer to any deploy pipeline that owns the instance right now. It is
// authoritative and will create or replace the actor itself.
if (InstanceIsOwnedByADeployPipeline(instanceName))
continue;
if (!_instanceActors.ContainsKey(instanceName))
{
// Row 78 — an enabled config with no actor.
if (!TryClaimSweepAttempt(instanceName, config.RevisionHash))
continue;
StartStoredInstance(config, SweepReason,
"An enabled instance had a deployed config in the site store but no running "
+ "Instance Actor. Configs that arrive by LocalDb replication (a standby "
+ "seeded from its peer, or a snapshot resync) are written after the "
+ "boot-time load and are invisible to the startup reconcile, so nothing "
+ "else would have started it.");
continue;
}
// Row 72 — a running actor serving a revision the store has replaced.
TryRestartStaleInstance(config);
}
}
/// <summary>
/// Claims the single sweep repair attempt for an (instance, revision) pair. Returns
/// <see langword="false"/> when this exact revision has already been attempted — see
/// <see cref="_sweepAttemptedRevisions"/> for why a periodic self-heal must not retry the
/// same repair forever.
/// </summary>
private bool TryClaimSweepAttempt(string instanceName, string revisionHash)
{
if (_sweepAttemptedRevisions.TryGetValue(instanceName, out var attempted) &&
string.Equals(attempted, revisionHash, StringComparison.Ordinal))
{
return false;
}
_sweepAttemptedRevisions[instanceName] = revisionHash;
return true;
}
/// <summary>
/// Deferred register row 72 — <b>the running Instance Actor is serving a config the durable
/// store has replaced.</b>
///
/// <para>
/// The startup reconcile refreshes a STALE row via <c>StoreDeployedConfigIfNewerAsync</c>
/// (and a CDC-replicated deploy lands the same way), which heals the file while the running
/// actor keeps executing the previously compiled config until the next restart or redeploy.
/// Nothing surfaced the difference.
/// </para>
///
/// <para>
/// <b>The decision (2026-08-15, <c>docs/plans/2026-08-15-instance-sweep-self-heal-design.md</c>):
/// converge, do not merely warn.</b> The stored row is what central successfully deployed;
/// the running actor is executing a config central believes was replaced. Serving it
/// indefinitely is the actual fault. The stored config is therefore applied through the
/// EXACTLY the same path a central redeploy uses — <see cref="HandleDeploy"/>, so the
/// compile gate, the warm-then-gate ordering, the stop/watch/recreate handshake, the
/// static-override and native-alarm reset, and all-or-nothing-per-instance are the ones
/// already in production, not a second implementation. A stop still does not cancel an
/// in-flight script run, exactly as for a central redeploy.
/// </para>
///
/// <para>
/// <b>Safety.</b> The restart is attempted once per stored revision, so a config the
/// compile gate rejects (which leaves the running actor untouched — the gate returns before
/// anything is stopped) is announced but never looped on. The sweep is also skipped
/// entirely while a deploy pipeline owns the instance, so it can never interleave with a
/// real redeploy.
/// </para>
/// </summary>
private void TryRestartStaleInstance(DeployedInstance config)
{
var instanceName = config.InstanceUniqueName;
var storedRevision = config.RevisionHash;
// No stored revision to compare against — nothing can be concluded, so do nothing.
if (string.IsNullOrEmpty(storedRevision))
return;
// No recorded running revision. Unreachable while CreateInstanceActor is the single
// creation funnel; treated as "unknown", never as "stale", so a future path that
// bypasses the funnel degrades to no-op rather than to spurious restarts.
if (!_runningRevisions.TryGetValue(instanceName, out var runningRevision) ||
string.IsNullOrEmpty(runningRevision))
{
return;
}
if (string.Equals(runningRevision, storedRevision, StringComparison.Ordinal))
return;
if (!TryClaimSweepAttempt(instanceName, storedRevision))
return;
_logger.LogWarning(
"Instance {Instance} is running stale revision {RunningRevision} while the deployed-config "
+ "store holds {StoredRevision} — applying the stored config through the redeploy path "
+ "({Reason} self-heal)",
instanceName, runningRevision, storedRevision, SweepReason);
LogDeploymentEvent("Warning", instanceName,
$"Instance {instanceName} restarted by {SweepReason} self-heal — "
+ $"running revision {runningRevision} is stale, applying stored revision {storedRevision}",
"The site's deployed-config store was refreshed out of band (startup reconcile "
+ "refreshing a stale row, or a replicated deploy) while the Instance Actor kept "
+ "serving the previously compiled config. The stored config is what central "
+ "successfully deployed, so it is applied through the normal redeploy path.");
// Reply to Self: this actor is its own deployer here, and HandleSweepRestartResult
// surfaces the outcome. Replying to NoSender would dead-letter it, and dead letters are
// a monitored site health metric.
HandleDeploy(
new DeployInstanceCommand(
config.DeploymentId,
instanceName,
storedRevision,
config.ConfigJson,
$"{SweepReason} self-heal",
DateTimeOffset.UtcNow),
Self);
}
/// <summary>
/// Terminal outcome of a sweep-driven stale-revision restart. Only the sweep makes this
/// actor its own deployer, so any <see cref="DeploymentStatusResponse"/> arriving here is
/// its own. A failure is surfaced and NOT retried (see
/// <see cref="_sweepAttemptedRevisions"/>); the instance keeps serving whatever config it
/// had, which for a compile-gate rejection is the old one, untouched.
/// </summary>
private void HandleSweepRestartResult(DeploymentStatusResponse response)
{
if (response.Status == DeploymentStatus.Success)
{
_logger.LogInformation(
"Instance {Instance} re-applied its stored config after a stale-revision "
+ "{Reason} self-heal (deploymentId={DeploymentId})",
response.InstanceUniqueName, SweepReason, response.DeploymentId);
return;
}
_logger.LogWarning(
"Stale-revision {Reason} self-heal for {Instance} did not apply: {Error}",
SweepReason, response.InstanceUniqueName, response.ErrorMessage);
LogDeploymentEvent("Error", response.InstanceUniqueName,
$"Instance {response.InstanceUniqueName} stale-revision {SweepReason} self-heal failed",
response.ErrorMessage ?? "no error detail");
}
/// <summary>
/// Deletes an instance: stops the actor and removes config from SQLite.
/// Note: store-and-forward messages are NOT cleared per design decision.
/// </summary>
private void HandleDelete(DeleteInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
if (DeferLifecycleStop(instanceName, command.CommandId, true, replyTo)) return;
// 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);
// Delete is the one lifecycle event after which the instance name never comes back on
// its own, so drop the sweep's per-instance bookkeeping with it rather than letting it
// accumulate for the process lifetime.
_runningRevisions.Remove(instanceName);
_sweepAttemptedRevisions.Remove(instanceName);
UpdateInstanceCounts();
var sender = replyTo;
// Row 101 (initiation half): the actor has already forgotten the instance (actor stopped,
// _deployedInstanceNames/_runningRevisions/_sweepAttemptedRevisions cleared) while the
// durable row is still there — and clearing _sweepAttemptedRevisions specifically hands
// the sweep a fresh repair claim, so an in-flight read landing now would resurrect the
// instance the operator just deleted.
BumpStoreWriteGeneration("delete dispatch");
var deleteWrite = _storage.RemoveDeployedConfigAsync(instanceName);
// Row 101 (completion half) — see StoreWriteCompleted.
deleteWrite.ContinueWith(_ => new StoreWriteCompleted("delete")).PipeTo(Self);
deleteWrite.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
// 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);
}
/// <summary>
/// Fire-and-forget a <c>deployment</c> operational event to the optional
/// <see cref="ISiteEventLogger"/> 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 / script-run
/// pattern).
/// <para>
/// <b>Thread-safety:</b> the disable (<see cref="HandleDisable"/>) and delete
/// (<see cref="HandleDelete"/>) paths call this from a
/// <see cref="System.Threading.Tasks.Task.ContinueWith(System.Action{System.Threading.Tasks.Task})"/>
/// 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 <c>readonly _serviceProvider</c> captured at
/// construction (the resolved <see cref="ISiteEventLogger"/> is a process
/// singleton). No actor-private mutable state is referenced, which is what
/// makes calling it off the actor thread safe.
/// </para>
/// </summary>
private void LogDeploymentEvent(string severity, string instanceName, string message, string? details = null)
{
_ = _serviceProvider?.GetService<ISiteEventLogger>()?.LogEventAsync(
"deployment", severity, instanceName, "DeploymentManagerActor", message, details);
}
/// <summary>
/// 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 <c>IsDeployed = false</c> with
/// null identity so central falls through to a normal deploy.
/// </summary>
private void HandleDeploymentStateQuery(DeploymentStateQueryRequest request)
{
var sender = Sender;
var instanceName = request.InstanceUniqueName;
// Row P1-4: a single-row, ConfigJson-free projection — this used to materialize every
// deployed config (including every ConfigJson blob) just to FirstOrDefault this one
// instance's DeploymentId/RevisionHash. Neither is ever read here.
_storage.GetDeployedConfigMetadataAsync(instanceName).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;
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 ──
/// <summary>
/// The base cluster role every site node carries (in addition to its
/// per-site role <c>site-{SiteId}</c>). 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 <c>NodeOptions.Role</c>
/// for site hosts (see <c>AkkaHostedService.BuildRoles</c>).
/// </summary>
private const string SiteClusterRole = "Site";
/// <summary>
/// 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).
/// </summary>
private static readonly TimeSpan CertBroadcastTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Broadcasts a trust to the per-node <see cref="CertStoreActor"/>
/// on every Up site node so both PKI stores receive the certificate.
/// </summary>
private void HandleTrustServerCert(TrustServerCertCommand command) =>
BroadcastToSiteCertStores(
new WriteCertToLocalStore(command.DerBase64, command.Thumbprint),
$"trust cert {command.Thumbprint} (connection {command.ConnectionName})");
/// <summary>
/// Broadcasts a remove to the per-node <see cref="CertStoreActor"/>
/// on every Up site node so the certificate leaves both PKI stores.
/// </summary>
private void HandleRemoveServerCert(RemoveServerCertCommand command) =>
BroadcastToSiteCertStores(
new RemoveCertFromLocalStore(command.Thumbprint),
$"remove cert {command.Thumbprint}");
/// <summary>
/// Lists this node's trusted + rejected PKI stores by asking the LOCAL
/// <see cref="CertStoreActor"/> (the singleton's own node). The list reflects
/// the active node's view; a trust broadcast keeps the standby in sync.
/// </summary>
private void HandleListServerCerts(ListServerCertsCommand command)
{
var sender = Sender;
var local = Context.ActorSelection($"/user/{CertStoreActor.WellKnownName}");
local.Ask<LocalCertOpAck>(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);
}
/// <summary>
/// Fans <paramref name="localMessage"/> out to the <see cref="CertStoreActor"/>
/// on every Up site node, asks each with a short timeout, and aggregates the
/// acks into one <see cref="CertTrustResult"/>. 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.
/// </summary>
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<LocalCertOpAck>(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);
}
/// <summary>
/// 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).
/// </summary>
private Address? _pendingCertReconcileTarget;
/// <summary>
/// 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.
/// </summary>
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));
}
/// <summary>
/// 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 <see cref="HandleLocalCertExport"/>, which uses the target captured here.
/// </summary>
private void HandleSiteNodeJoined(SiteNodeJoined msg)
{
_pendingCertReconcileTarget = msg.Address;
Context.ActorSelection($"/user/{CertStoreActor.WellKnownName}")
.Tell(new ExportLocalTrustedCerts());
}
/// <summary>
/// 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.
/// </summary>
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 ──
/// <summary>
/// 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 that is new gets a
/// <c>CreateConnectionCommand</c>, and a name whose config CHANGED gets a
/// <c>ReconfigureConnectionCommand</c> so the live connection actor adopts the new
/// configuration in place instead of serving a stale one until it is next rebuilt.
/// </summary>
private readonly Dictionary<string, string> _createdConnections = new();
/// <summary>
/// <b>The DCL connection arbitration overlay</b> (deferred register rows 89 + 90) — the
/// <c>data_connection_definitions</c> rows this singleton incarnation currently believes the
/// site store owns, keyed by connection name.
///
/// <para>
/// <b>The rule: the definitions store wins wherever it has an opinion.</b> A connection name
/// with a row here is configured from that row; the identically-named entry inside an
/// instance's flattened config is a deploy-time SNAPSHOT and never overrides it. A name with
/// no row here falls through to the flattened config exactly as before — that is not a
/// fallback, it is the correct answer for a deploy-only site, where this table is legitimately
/// EMPTY because it is written only by the system-wide artifact deploy while the connections
/// that matter live inline in the instance configs.
/// </para>
///
/// <para>
/// <b>This overlay is an arbiter, not a creation source.</b> A definition row for a name no
/// deployed instance references creates nothing: there is no connection actor wanted for it.
/// The overlay only decides WHICH configuration a connection the runtime already wants is
/// created from — which is what makes it safe to consult on every sweep tick.
/// </para>
///
/// <para>
/// Refreshed by <see cref="ApplyConnectionDefinitionOverlay"/> from the boot load and from
/// every sweep tick, and kept consistent in-flight by
/// <see cref="HandleApplyArtifactDataConnectionsToDcl"/>. Actor-thread-confined.
/// </para>
/// </summary>
private readonly Dictionary<string, StoredDataConnectionDefinition> _connectionDefinitions =
new(StringComparer.Ordinal);
/// <summary>
/// Connection names this singleton incarnation has seen REMOVED centrally (deferred register
/// row 90) — either by an artifact reconcile it processed itself, or by a definition row that
/// disappeared from the store between two sweep reads.
///
/// <para>
/// Without it the removal does not stick for as long as one tick: the removal evicts the name
/// from <see cref="_createdConnections"/>, and the very next instance start whose flattened
/// config still names that connection re-creates it — the flattened snapshot resurrecting a
/// connection central has deleted. That is not hypothetical; the sweep starts instances.
/// </para>
///
/// <para>
/// <b>Deliberately in-memory only.</b> Persisting tombstones was rejected: a durable
/// "this connection is gone" marker is a second, divergent authority over the same names, and
/// the flattened configs already resurrect a removed connection across a restart today — that
/// pre-existing semantic is accepted and unchanged, and is recorded as the residual in
/// <c>docs/plans/2026-08-17-dcl-connection-arbitration-decision.md</c>. Cleared the moment a
/// definition row for the name reappears, so a centrally re-added connection comes straight
/// back.
/// </para>
/// </summary>
private readonly HashSet<string> _centrallyRemovedConnections = new(StringComparer.Ordinal);
/// <summary>
/// Monotonic count of this actor's own mutations of the two tables
/// <see cref="ReadStoreSnapshotAsync"/> reads — <c>deployed_configurations</c> AND
/// <c>data_connection_definitions</c> (deferred register row 101, extended to the config
/// half). A store snapshot is read OFF the actor thread but consumed only when its piped
/// message lands, so a read dispatched before a mutation can land after the actor's own
/// state has already advanced past it. Every consumer therefore captures this value on the
/// actor thread at read dispatch and discards the WHOLE snapshot on a mismatch.
///
/// <para>
/// <b>One counter for both tables, deliberately.</b> The snapshot is one consistent read and
/// is discarded whole, so a per-table counter would buy nothing but the chance to apply half
/// a snapshot that the other half's mutation already invalidated.
/// </para>
///
/// <para>
/// <b>What a stale snapshot would actually do</b> — each of these is a real defect the guard
/// closes, not a theoretical one:
/// <list type="bullet">
/// <item>DEFINITIONS half: absences "prove" just-applied names disappeared, tombstoning and
/// STOPPING live connections that nothing re-creates until an instance redeploy; rows push
/// superseded configs onto live connections and flip-flop with the next fresh read.</item>
/// <item>CONFIGS half, row-72 leg: a snapshot carrying the revision a deploy has just
/// replaced makes <see cref="TryRestartStaleInstance"/> "repair" the instance BACKWARD —
/// it redeploys the snapshot's own (old) config JSON through the redeploy path.
/// <see cref="InstanceIsOwnedByADeployPipeline"/> does not cover this: it only holds while
/// the deploy is in flight, and the stale snapshot can land after the pipeline released
/// ownership.</item>
/// <item>CONFIGS half, row-78 leg: a snapshot read before an undeploy/delete or a disable
/// committed still shows an enabled row while the actor is already stopped, so
/// <see cref="StartStoredInstance"/> RESURRECTS a just-stopped instance.</item>
/// </list>
/// </para>
///
/// <para>
/// <b>Bumped twice per mutation — at initiation and at completion</b> (see
/// <see cref="BumpStoreWriteGeneration"/>), which is what makes the guard airtight rather
/// than merely usually-right. The store write runs off the actor thread, so a read
/// dispatched between "the actor decided to mutate" and "the write committed" would compare
/// EQUAL against an initiation-only bump and be applied even though it predates the write.
/// Bumping again on the completion message — which is created only after the write has
/// committed — makes any such in-window read compare stale. Consequently: a read dispatched
/// before the mutation is discarded, a read dispatched during it is discarded, and a read
/// dispatched after the completion bump necessarily sees the committed row.
/// </para>
///
/// <para>
/// <b>Writes that must NOT bump.</b> Rows arriving by LocalDb CDC replication, by a snapshot
/// resync, by an operator restoring the file, or by <c>SiteReconciliationActor</c>'s
/// startup fetch never pass through this actor at all — and discovering exactly those is the
/// instance reconcile sweep's whole reason to exist (register rows 72 + 78). They are not
/// "stale relative to actor state"; they are the state the actor has yet to learn about.
/// Bumping for them (were it even possible) would blind the sweep to its own use case.
/// </para>
/// </summary>
private long _storeWriteGeneration;
/// <summary>
/// Records that this actor has mutated one of the two snapshot tables, invalidating every
/// store read currently in flight. Called on the actor thread at BOTH ends of a mutation —
/// when the off-thread write is dispatched and again when its completion message lands. See
/// <see cref="_storeWriteGeneration"/> for why both ends are required.
/// </summary>
/// <param name="source">Provenance for the Trace line; never affects behaviour.</param>
private void BumpStoreWriteGeneration(string source)
{
_storeWriteGeneration++;
_logger.LogTrace(
"Store write generation bumped to {Generation} by {Source}", _storeWriteGeneration, source);
}
/// <summary>
/// The row-101 freshness test both snapshot consumers run BEFORE touching anything the
/// snapshot carries. <see langword="true"/> means this actor mutated one of the two tables
/// while the read was in flight, so the snapshot predates state the actor has already
/// advanced past and must be discarded whole (see <see cref="_storeWriteGeneration"/> for
/// what each half would otherwise do). Logged at Debug: on a healthy site this fires only
/// when a deploy/enable/disable/delete happens to overlap a sweep tick, which is normal and
/// self-correcting.
/// </summary>
private bool IsStoreSnapshotStale(long generationAtRead, string source)
{
if (generationAtRead == _storeWriteGeneration)
return false;
_logger.LogDebug(
"Discarding the {Source} store snapshot — this node mutated the deployed configs or "
+ "the connection definitions while the read was in flight (generation "
+ "{ReadGeneration} vs {CurrentGeneration})",
source, generationAtRead, _storeWriteGeneration);
return true;
}
/// <summary>
/// Refreshes <see cref="_connectionDefinitions"/> from a store read and reconciles the live
/// DCL against it (deferred register rows 89 + 90).
///
/// <para>
/// <paramref name="definitions"/> is <see langword="null"/> when the read FAILED, which is
/// deliberately distinct from an empty list: a failed read leaves the overlay untouched,
/// while an empty list is a real answer (the deploy-only site shape, and also what a site
/// looks like after central deletes every data connection).
/// </para>
///
/// <para>
/// <b>Disappearance detection, and exactly how far it reaches.</b> A name that WAS in this
/// overlay and no longer has a row is a removal this incarnation can prove: the store owned
/// it a moment ago and does not now. It is tombstoned and its live connection actor stopped.
/// A name that was never in the overlay is indistinguishable from one the store has never
/// owned — a fresh incarnation cannot tell "central deleted this" from "this site keeps its
/// connections inline" — so it is left to the flattened config. That asymmetry is the
/// documented residual of rows 89/90, not an oversight.
/// </para>
///
/// <para>
/// Reached only through <see cref="ApplyConnectionDefinitionOverlayIfFresh"/> (row 101) —
/// never with a snapshot proven stale.
/// </para>
/// </summary>
private void ApplyConnectionDefinitionOverlay(
IReadOnlyList<StoredDataConnectionDefinition>? definitions)
{
if (definitions == null)
return;
var disappeared = _connectionDefinitions.Keys
.Where(name => !definitions.Any(d => string.Equals(d.Name, name, StringComparison.Ordinal)))
.ToList();
foreach (var definition in definitions)
{
// A definition row is the store re-asserting ownership: any tombstone for the name
// is stale, so a centrally re-added connection is creatable again immediately.
_centrallyRemovedConnections.Remove(definition.Name);
_connectionDefinitions[definition.Name] = definition;
// Push only to connections the runtime already holds. A definition nothing
// references stays inert — see the overlay's "arbiter, not a creation source" note.
if (_createdConnections.ContainsKey(definition.Name))
{
EnsureDclConnection(
definition.Name,
definition.Protocol,
definition.ConfigurationJson,
definition.BackupConfigurationJson,
definition.FailoverRetryCount);
}
}
foreach (var name in disappeared)
RemoveConnectionCentrally(name, "its definition row is no longer in the site store");
}
/// <summary>
/// Applies a central removal of a data connection: tombstones the name for this singleton
/// incarnation, drops the overlay row and the hash-cache slot, and stops the live DCL
/// connection actor. Shared by the artifact-reconcile removal and the sweep's
/// disappearance detection so both produce identical state.
///
/// <para>
/// A connection still referenced by a deployed instance's config no longer exists centrally
/// either — its attributes go bad quality, the standard disconnected signal, until the
/// instance is redeployed against current central config. That is the pre-existing removal
/// semantic and it is unchanged here; what IS new is that the removal now survives the next
/// instance start instead of being undone by it.
/// </para>
/// </summary>
private void RemoveConnectionCentrally(string name, string because)
{
_connectionDefinitions.Remove(name);
_centrallyRemovedConnections.Add(name);
_createdConnections.Remove(name);
_dclManager?.Tell(new DataConnectionLayer.Actors.RemoveConnectionCommand(name));
_logger.LogWarning(
"Data connection '{Name}' removed centrally ({Because}); its DCL connection actor "
+ "has been stopped and will not be recreated from an instance's flattened config",
name, because);
}
/// <summary>
/// Sets up DCL connections from the flattened config, <b>arbitrated against
/// <see cref="_connectionDefinitions"/></b> (deferred register rows 89 + 90). Idempotent on
/// unchanged configuration.
///
/// <para>
/// <b>Why the arbitration has to happen HERE, before the hash is computed.</b>
/// <see cref="_createdConnections"/> holds one hash slot per connection name. The flattened
/// snapshot and the definition row for the same name can legitimately differ — that is the
/// supported version-skew case, an artifact deploy newer than the instance deploys that
/// reference it. Two callers hashing from two different sources would flip that one slot back
/// and forth, and every flip is a fresh <c>CreateConnectionCommand</c>. Resolving the WINNER
/// first means only one configuration can ever be hashed for a name, whichever path calls, so
/// the flip is not merely avoided by convention — it is unrepresentable.
/// </para>
/// </summary>
private void EnsureDclConnections(string configJson)
{
if (_dclManager == null) return;
try
{
var config = System.Text.Json.JsonSerializer.Deserialize<Commons.Types.Flattening.FlattenedConfiguration>(configJson);
if (config?.Connections == null) return;
foreach (var (name, connConfig) in config.Connections)
{
// Row 90: never resurrect a name the artifact reconcile removed.
if (_centrallyRemovedConnections.Contains(name))
{
_logger.LogDebug(
"Skipping DCL connection '{Connection}' named by a flattened config — it "
+ "was removed centrally in this incarnation", name);
continue;
}
if (_connectionDefinitions.TryGetValue(name, out var owned))
{
// Rule 1: the store has an opinion, so the store wins.
EnsureDclConnection(
name,
owned.Protocol,
owned.ConfigurationJson,
owned.BackupConfigurationJson,
owned.FailoverRetryCount);
continue;
}
// Rule 2: no definition row has ever owned this name — the flattened config is
// the source, exactly as before. This is the whole of a deploy-only site.
EnsureDclConnection(
name,
connConfig.Protocol,
connConfig.ConfigurationJson,
connConfig.BackupConfigurationJson,
connConfig.FailoverRetryCount);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to parse flattened config for DCL connections");
}
}
/// <summary>
/// Hash-guarded DCL connection push shared by the inline
/// per-instance path (<see cref="EnsureDclConnections(string)"/>) and the
/// system-wide artifact-deploy path (<see cref="HandleDeployArtifacts"/>).
///
/// <para>
/// <b>Three outcomes, one hash slot.</b> An unchanged configuration is a no-op (no message,
/// no log — this is what keeps a healthy site silent across every sweep tick). A name the
/// hash cache has never held gets a <c>CreateConnectionCommand</c>. A name whose
/// endpoint/credentials/backup/failover-count CHANGED gets a
/// <c>ReconfigureConnectionCommand</c>, which the DCL applies to the live connection actor
/// <b>in place</b> — it swaps its adapter through the same machinery the primary→backup
/// failover uses and re-issues every subscription it holds.
/// </para>
///
/// <para>
/// <b>Why the changed case is not a re-issued create</b> (deferred register row 97).
/// <c>DataConnectionManagerActor.HandleCreateConnection</c> is CREATE-ONLY — it warns and
/// returns for a name it already holds — so the re-issued create this path used to send
/// never reached the live actor at all, while the hash cache recorded it as applied. A
/// changed endpoint therefore took effect only when the connection was next built from
/// scratch (node restart, singleton failover, or a remove-then-add). Stopping and
/// re-creating the actor is not the fix either: subscriptions live inside the connection
/// actor and <c>InstanceActor</c> cancels its subscribe-retry timer on first success, so
/// nothing would re-subscribe. See
/// <c>docs/plans/2026-08-17-dcl-connection-arbitration-decision.md</c> §9.
/// </para>
/// </summary>
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);
var changed = _createdConnections.ContainsKey(name);
_createdConnections[name] = configHash;
if (changed)
{
_dclManager.Tell(new Commons.Messages.DataConnection.ReconfigureConnectionCommand(
name, protocol, primaryDetails, backupDetails, failoverRetryCount));
// Emitted once per changed configuration (the hash guard above suppresses repeats),
// so a healthy site stays silent.
_logger.LogInformation(
"Reconfigured DCL connection {Connection} (protocol={Protocol}) in place — the "
+ "live connection actor swaps its adapter onto the new configuration and "
+ "re-issues its subscriptions; no redeployment or restart is needed",
name, protocol);
return;
}
_dclManager.Tell(new Commons.Messages.DataConnection.CreateConnectionCommand(
name, protocol, primaryDetails, backupDetails, failoverRetryCount));
_logger.LogInformation(
"Created DCL connection {Connection} (protocol={Protocol})", name, protocol);
}
/// <summary>
/// Computes a stable hash over the configuration fields that affect how the DCL
/// connects, so a changed endpoint/credential/backup/failover count is detected.
/// </summary>
private static string ComputeConnectionConfigHash(
Commons.Types.Flattening.ConnectionConfig connConfig) =>
ComputeConnectionConfigHashCore(
connConfig.Protocol,
connConfig.ConfigurationJson,
connConfig.BackupConfigurationJson,
connConfig.FailoverRetryCount);
/// <summary>
/// Field-based core so the system-wide artifact-deploy
/// path (which carries protocol/config-json/backup-json/failover directly
/// on <see cref="Commons.Messages.Artifacts.DataConnectionArtifact"/>) can
/// share the same hash + skip-or-resend logic as the inline-config path.
/// </summary>
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<string, string> FlattenConnectionConfig(string protocol, string? json)
{
if (string.IsNullOrEmpty(json))
return new Dictionary<string, string>();
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<string, string>();
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<string, string>();
}
}
// ── Shared Script Loading ──
/// <summary>
/// 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 <see cref="SharedScriptsLoaded"/> message, carrying the
/// enabled configs to resume staggered Instance Actor creation on the actor thread.
/// </summary>
private void LoadSharedScriptsFromStorage(List<DeployedInstance> 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<Commons.Messages.Streaming.AttributeValueChanged>(),
Array.Empty<Commons.Messages.Streaming.AlarmStateChanged>(), 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<Commons.Messages.Streaming.AttributeValueChanged>(),
Array.Empty<Commons.Messages.Streaming.AlarmStateChanged>(), 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,
// Flow Debugger (design 2026-09-05 §5.2): the trace flag rides beside the parent id.
Trace: request.Trace);
var sender = Sender;
instanceActor.Ask<ScriptCallResult>(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,
// Flow Debugger (design §5.4): the site run's id and, for a traced
// run, its ScriptTrace events — returned inline so a Live run is
// instant. Null from an untraced run.
result.ExecutionId, result.TraceEvents);
}
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));
}
}
/// <summary>
/// Projects a routed script's return value to a cross-process-serializable plain
/// CLR graph (<see cref="Dictionary{TKey,TValue}"/> / <see cref="List{T}"/> /
/// string / long / double / bool / null) via a JSON round-trip. A script's natural
/// <c>return new { ... }</c> is a compiler-generated anonymous type that Akka's
/// cross-process serializer cannot reconstruct on the receiving node, so the
/// <see cref="RouteToCallResponse"/> 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 <paramref name="value"/>
/// unchanged on a (non-expected) serialization failure so a quirky value still has a
/// chance to deliver rather than being forced to null.
/// </summary>
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;
}
}
/// <summary>
/// Converts a <see cref="System.Text.Json.JsonElement"/> to a plain CLR value
/// (string / long / double / bool / null, or nested Dictionary / List) — never a
/// <see cref="System.Text.Json.JsonElement"/> (which is itself not cross-process
/// serializable). Companion to <see cref="NormalizeRoutedReturnValue"/>.
/// </summary>
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,
};
/// <summary>
/// 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.
/// </summary>
private void RouteInboundApiGetAttributes(RouteToGetAttributesRequest request)
{
if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
{
Sender.Tell(new RouteToGetAttributesResponse(
request.CorrelationId, new Dictionary<string, object?>(), 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<GetAttributeResponse>(
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<string, object?>();
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<T> — 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<string, object?>(), false,
t.Exception?.GetBaseException().Message ?? "Attribute read timed out",
DateTimeOffset.UtcNow);
}).PipeTo(sender);
}
/// <summary>
/// Spec §6 (WD-2b): unpacks a routed <see cref="RouteToWaitForAttributeRequest"/>
/// (inbound-API <c>Route.To().WaitForAttribute()</c>) into the deployed
/// Instance Actor's site-local <see cref="WaitForAttributeRequest"/> and relays
/// the result back. Value-equality across the wire — the predicate is null (a
/// cross-process lambda cannot be routed), but <c>RequireGoodQuality</c> is honored
/// from <see cref="RouteToWaitForAttributeRequest.RequireGoodQuality"/>. The Ask is
/// bounded by the wait timeout plus slack (NOT a fixed 30s), since the wait
/// legitimately blocks for up to <see cref="RouteToWaitForAttributeRequest.Timeout"/>.
/// </summary>
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<WaitForAttributeResponse>(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<T>
// (List<int>/List<double>/List<DateTime>/… 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);
}
/// <summary>
/// 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.
/// </summary>
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<SetStaticAttributeResponse>(
new SetStaticAttributeCommand(
correlationId, request.InstanceUniqueName, kvp.Key, kvp.Value, DateTimeOffset.UtcNow,
// (ParentExecutionId tag-cascade): the routed
// write's originating execution is the inbound API request
// that issued Route.To(...).SetAttributes(...) — already
// carried on the request as ParentExecutionId. Stamping it
// here lets an alarm this write trips chain its on-trigger
// script run under the inbound execution.
SourceExecutionId: request.ParentExecutionId,
// Flow Debugger (design §5.3): a traced routed write arms the echo stamp.
Trace: request.Trace),
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);
}
/// <summary>
/// Handles system-wide artifact deployment (shared scripts, external systems, etc.).
/// Persists artifacts to SiteStorageService and recompiles shared scripts.
/// </summary>
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;
// Row 101 (initiation half): the artifact apply rewrites data_connection_definitions
// off-thread. The completion half is the ApplyArtifactDataConnectionsToDcl handler; this
// bump covers the window before it, where a read would otherwise land carrying the
// pre-apply table and be treated as current.
BumpStoreWriteGeneration("artifact apply dispatch");
Task.Run(async () =>
{
try
{
// Each non-null artifact list is the COMPLETE system-wide set
// (ArtifactDeploymentService ships GetAll* snapshots), so the apply is
// upsert-then-reconcile: store every incoming artifact, then delete any
// stored row absent from the set — that artifact was deleted centrally.
// Upsert-only apply used to leave such rows orphaned on the site forever
// (a centrally-deleted external system stayed callable from site scripts).
// A null list means "field not shipped" and touches nothing.
// 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);
}
var removedScripts = await _storage.DeleteSharedScriptsExceptAsync(
command.SharedScripts.Select(s => s.Name).ToList());
foreach (var name in removedScripts)
_sharedScriptLibrary.Remove(name);
}
// 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,
es.MaxRetries, es.RetryDelay);
}
await _storage.DeleteExternalSystemsExceptAsync(
command.ExternalSystems.Select(es => es.Name).ToList());
// WP2.6a: drop the shared read cache so the next script call sees
// the freshly-applied definitions/method lists instead of a stale
// snapshot loaded before this deploy.
_externalSystemCache?.InvalidateAll();
}
// 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);
}
await _storage.DeleteDatabaseConnectionsExceptAsync(
command.DatabaseConnections.Select(db => db.Name).ToList());
}
// 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);
}
var removedConnections = await _storage.DeleteDataConnectionDefinitionsExceptAsync(
command.DataConnections.Select(dc => dc.Name).ToList());
// 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. Reconciled removals ride the same
// message so the live DCL connection actor is stopped
// on the actor thread alongside the hash-cache eviction.
self.Tell(new ApplyArtifactDataConnectionsToDcl(
command.DataConnections, removedConnections));
}
// SMTP configuration is
// central-only and is never stored on a site (see the purge above).
// Site-secrets artifact deploy (Task 10): write the shipped secrets to the
// replicated deployed_secrets table, then reconcile THIS node's own mirror
// synchronously — the reply's SecretsApplied must reflect this node's own
// apply/verify outcome, not whatever the next periodic pass finds. A null
// command.Secrets means "central didn't ship secrets" (old-central
// compatible) and touches nothing; a non-null Secrets with no mirror
// registered on this node fails the deploy rather than silently applying
// nothing.
int? secretsApplied = null;
string? secretError = null;
if (command.Secrets != null)
{
if (_secretsMirror == null)
{
if (command.Secrets.Count > 0)
{
secretError = "Secret deployment is not enabled on this site node";
}
else
{
secretsApplied = 0;
}
}
else
{
_logger.LogInformation(
"Applying {Count} deployed secret(s), deploymentId={DeploymentId}",
command.Secrets.Count, command.DeploymentId);
await _storage.UpsertDeployedSecretsAsync(command.Secrets, command.DeploymentId);
var mirror = await _secretsMirror.ReconcileAsync(CancellationToken.None);
// The mirror reconciles the WHOLE site-wide deployed_secrets table
// every pass — not only the names THIS deploy just shipped — so a
// failure unrelated to this deploy's own secrets (another instance's
// already-deployed secret still failing, or the removal of a
// no-longer-referenced one) must never fail THIS response or show up
// in its ErrorMessage. Only failures among the SHIPPED names count.
var shippedNames = new HashSet<string>(
command.Secrets.Select(s => s.Name), StringComparer.Ordinal);
var shippedFailureMessages = new List<string>();
var shippedFailedNames = new HashSet<string>(StringComparer.Ordinal);
for (var i = 0; i < mirror.FailedNames.Count; i++)
{
var failedName = mirror.FailedNames[i];
var failedMessage = mirror.Failed[i];
if (shippedNames.Contains(failedName))
{
shippedFailedNames.Add(failedName);
shippedFailureMessages.Add(failedMessage);
}
else
{
_logger.LogWarning(
"Deployed secrets mirror reported a failure unrelated to " +
"this deploy ({DeploymentId}): {Failure}",
command.DeploymentId, failedMessage);
}
}
// Never negative (ruling): shippedFailedNames is a set intersected
// against the shipped names, so this can only go below zero if the
// shipped list itself contains duplicate names.
secretsApplied = Math.Max(0, command.Secrets.Count - shippedFailedNames.Count);
if (shippedFailureMessages.Count > 0)
{
secretError = string.Join("; ", shippedFailureMessages);
}
}
}
// Replicate artifacts to standby node
return new ArtifactDeploymentResponse(
command.DeploymentId, "", secretError is null, secretError, DateTimeOffset.UtcNow,
secretsApplied);
}
catch (Exception ex)
{
return new ArtifactDeploymentResponse(
command.DeploymentId, "", false, ex.Message, DateTimeOffset.UtcNow);
}
}).PipeTo(sender);
}
/// <summary>
/// Creates a child Instance Actor with the given name and configuration JSON.
/// </summary>
/// <param name="instanceName">The unique name of the instance to create an actor for.</param>
/// <param name="configJson">The JSON-serialized flattened configuration for the instance.</param>
/// <param name="revisionHash">
/// Revision hash of <paramref name="configJson"/>, recorded in <see cref="_runningRevisions"/>
/// so the instance reconcile sweep can tell an actor serving the CURRENT stored config from
/// one still serving a revision the store has replaced (deferred register row 72). This is
/// the single creation funnel, so recording it here covers every path that can ever start an
/// Instance Actor.
/// </param>
/// <param name="staticStateResetGate">
/// The deploy path's static-override / native-alarm reset, dispatched BEFORE this call so the
/// new actor's PreStart override load can be ordered after it (arch-review 2026-08-19 P1-2).
/// <see langword="null"/> on every non-deploy start (boot batch, enable, self-heal), where
/// nothing is being reset and the load runs immediately as before.
/// </param>
internal void CreateInstanceActor(
string instanceName, string configJson, string revisionHash,
Task? staticStateResetGate = null)
{
// Deferred register row 85 — an Instance Actor is useless without the DCL connection
// actors its data-sourced attributes subscribe through, so ensuring them belongs in the
// single creation funnel rather than at each caller.
//
// This used to be the caller's job, and only TWO of the four callers did it: the
// boot-time batch and ApplyDeployment. The two self-heal starts did not —
// StartStoredInstance (the shared row-67 ensure-started / row-78 sweep path) and
// HandleEnableResult. Both therefore created an instance whose every tag subscribe was
// answered "Unknown connection: X" by the DCL manager, forever, because nothing else on
// the node would ever create that connection actor:
//
// - Row 85's live symptom (2026-08-16): a replication-seeded deployed_configurations row
// is started by the row-78 sweep, but the connection actor is only ever created as a
// side effect of a deploy or the boot load, so tag subscribes failed for 6+ minutes
// until a second pair restart let the boot load create it.
// - The same defect is reachable with no replication at all: the boot load reads only
// ENABLED configs, so a node that boots with an instance DISABLED never creates that
// instance's connections, and enabling it afterwards hit the identical dead end.
//
// Placed BEFORE the already-exists guard so the two paths that already called it keep
// their exact previous semantics (they called it unconditionally). It is cheap and
// silent when there is nothing to do: EnsureDclConnection is hash-guarded on
// _createdConnections, so an unchanged connection is a no-op with no message and no log.
EnsureDclConnections(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<InstanceActor>(),
_dclManager,
_healthCollector,
_serviceProvider,
null,
staticStateResetGate));
var actorRef = Context.ActorOf(props, instanceName);
_instanceActors[instanceName] = actorRef;
_runningRevisions[instanceName] = revisionHash;
// 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);
}
/// <summary>
/// Actor-thread handler that pushes artifact-deploy data
/// connection definitions to the DCL via the shared
/// <see cref="EnsureDclConnection"/> helper. Dispatched from
/// <see cref="HandleDeployArtifacts"/>'s off-thread Task so the
/// <see cref="_createdConnections"/> hash-cache mutation stays
/// actor-thread-confined.
/// </summary>
private void HandleApplyArtifactDataConnectionsToDcl(ApplyArtifactDataConnectionsToDcl msg)
{
// Row 101 (completion half): the definitions rows are committed by the time this message
// is dispatched, so any snapshot read that started earlier carries data predating them —
// mark it stale so it is discarded when it lands.
BumpStoreWriteGeneration("artifact apply");
foreach (var dc in msg.DataConnections)
{
// Rows 89 + 90: the rows have just been committed to data_connection_definitions, so
// update the arbitration overlay in the same pass rather than waiting for the next
// sweep read. Skipping this would leave a window in which an instance start resolves
// the name to its flattened snapshot while the artifact path has already pushed the
// definition — the exact two-sources-one-hash-slot flip the overlay exists to prevent.
_centrallyRemovedConnections.Remove(dc.Name);
_connectionDefinitions[dc.Name] = new StoredDataConnectionDefinition
{
Name = dc.Name,
Protocol = dc.Protocol,
ConfigurationJson = dc.PrimaryConfigurationJson,
BackupConfigurationJson = dc.BackupConfigurationJson,
FailoverRetryCount = dc.FailoverRetryCount
};
EnsureDclConnection(
dc.Name,
dc.Protocol,
dc.PrimaryConfigurationJson,
dc.BackupConfigurationJson,
dc.FailoverRetryCount);
}
// Reconciled removals: the definition row is already gone from SQLite; drop the overlay
// row, tombstone the name for this incarnation (row 90 — otherwise the next instance
// start recreates it from that instance's flattened config), evict the hash cache so a
// later re-add recreates cleanly, and stop the live DCL connection actor.
foreach (var name in msg.RemovedConnectionNames)
RemoveConnectionCentrally(name, "deleted centrally, reconciled by an artifact deploy");
}
/// <summary>
/// Gets the count of active Instance Actors (for testing/diagnostics).
/// </summary>
internal int InstanceActorCount => _instanceActors.Count;
/// <summary>
/// Updates the health collector with current instance counts, and (WP3.1) grows the
/// script-execution pool to match.
/// Total deployed = _deployedInstanceNames.Count, enabled = running actors, disabled = difference.
/// </summary>
private void UpdateInstanceCounts()
{
_healthCollector?.SetInstanceCounts(
deployed: _deployedInstanceNames.Count,
enabled: _instanceActors.Count,
disabled: _deployedInstanceNames.Count - _instanceActors.Count);
// WP3.1: the blocking script pool scales with the number of running instances rather
// than sitting at a fixed 8 forever. This is the single call site because it already
// runs on every deploy / undeploy / enable / disable and once per staggered startup
// batch. Growth is idempotent and one-way — see ScriptExecutionScheduler.EnsureCapacity.
try
{
var target = ScriptExecutionScheduler.ComputeTargetThreads(_instanceActors.Count, _options);
(_scriptScheduler ?? ScriptExecutionScheduler.Shared(_options)).EnsureCapacity(target);
}
catch (Exception ex)
{
// Pool sizing is an optimisation, never a correctness requirement: a failure here
// must not fail the deploy/enable/disable that triggered it.
_logger.LogWarning(ex, "Failed to resize the script-execution pool; continuing at its current size.");
}
}
// ── Internal messages ──
/// <summary>
/// UA1: a site cluster node has come Up and should receive a cert-trust
/// reconcile push. Derived from <see cref="ClusterEvent.MemberUp"/> (self and
/// non-site members already filtered out) so the reconcile flow is unit-testable
/// without constructing a real cluster membership event.
/// </summary>
internal sealed record SiteNodeJoined(Address Address);
/// <summary>
/// Carries the boot-time store read back onto the actor thread.
///
/// <para>
/// <paramref name="ConnectionDefinitions"/> (deferred register rows 89 + 90) is the
/// <c>data_connection_definitions</c> snapshot read in the SAME off-thread task, so the DCL
/// arbitration overlay is populated BEFORE the boot batch creates any Instance Actor — a
/// connection whose definition row is newer than the flattened snapshot inside the instance
/// config is therefore created from the definition on the very first push, not created stale
/// and corrected a sweep later. <see langword="null"/> means the definitions read FAILED and
/// the previous overlay must be kept; an EMPTY list means the table is genuinely empty, which
/// is the deploy-only site shape (see <see cref="ApplyConnectionDefinitionOverlay"/>).
/// </para>
/// </summary>
/// <param name="StoreGenerationAtRead">
/// Row 101: <see cref="_storeWriteGeneration"/> captured on the actor thread when the read
/// was dispatched. A mismatch on landing means this actor mutated the deployed configs or
/// the connection definitions while the read was in flight, so the WHOLE snapshot predates
/// the actor's own state. The boot load cannot simply drop it (startup has to finish), so
/// <see cref="HandleStartupConfigsLoaded"/> re-reads instead.
/// </param>
internal record StartupConfigsLoaded(
List<DeployedInstance> Configs,
string? Error,
IReadOnlyList<StoredDataConnectionDefinition>? ConnectionDefinitions = null,
long StoreGenerationAtRead = 0);
/// <summary>
/// Internal singleton signal that re-attempts the startup deployed-config load
/// after a transient SQLite failure (S5). Delivered by a single-shot Akka timer.
/// </summary>
internal sealed record RetryStartupLoad
{
/// <summary>The shared singleton instance.</summary>
public static readonly RetryStartupLoad Instance = new();
private RetryStartupLoad() { }
}
/// <summary>
/// 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.
/// </summary>
internal record SharedScriptsLoaded(
List<DeployedInstance> EnabledConfigs, int CompiledCount, int TotalCount);
/// <summary>
/// WP3.1 warm-then-gate: piped back to self once a deployment's scripts have been
/// compiled off the actor thread into the process-wide compile cache. Carries the exact
/// command instance it warmed so a result superseded mid-warm is recognised by reference
/// and dropped.
/// </summary>
internal sealed record DeployCompileWarmed(DeployInstanceCommand Command);
/// <summary>
/// WP3.1: piped back to self once a staggered startup batch's compiles have been
/// pre-warmed, carrying the original batch message so actor creation resumes unchanged.
/// </summary>
internal sealed record BatchCompileWarmed(StartNextBatch Batch);
/// <summary>
/// A lifecycle command queued because its instance's deploy was mid-compile-warm, with
/// the sender to answer once it is re-dispatched.
/// </summary>
internal sealed record BufferedInstanceCommand(object Command, IActorRef Sender);
/// <summary>
/// The in-flight state of one instance's deploy compile warm: the deploy that will be
/// applied when the warm lands (replaceable last-write-wins while nothing is queued behind
/// it), the deployer to answer, and the ordered commands queued during the warm.
/// </summary>
internal sealed class DeployWarmState(DeployInstanceCommand command, IActorRef replyTo)
{
/// <summary>The deploy to apply when the warm completes.</summary>
public DeployInstanceCommand Command { get; set; } = command;
/// <summary>The deployer awaiting this deploy's <see cref="DeploymentStatusResponse"/>.</summary>
public IActorRef ReplyTo { get; set; } = replyTo;
/// <summary>Commands for this instance that arrived during the warm, in arrival order.</summary>
public List<BufferedInstanceCommand> Buffered { get; } = [];
}
internal record StartNextBatch(BatchState State);
internal record BatchState(List<DeployedInstance> Configs, int NextIndex);
internal record EnableResult(
EnableInstanceCommand Command, DeployedInstance? Config, string? Error, IActorRef OriginalSender);
/// <summary>
/// Carries the durable row read for an <see cref="EnsureInstanceStartedCommand"/> back onto
/// the actor thread. No original sender: the ensure-started path is fire-and-forget, so
/// there is nobody to answer.
/// </summary>
/// <param name="StoreGenerationAtRead">
/// Row 101: <see cref="_storeWriteGeneration"/> captured on the actor thread when the read
/// was dispatched. A mismatch on landing means a disable/delete/deploy committed while the
/// read was in flight, and the heal is dropped whole (see
/// <see cref="HandleEnsureInstanceStartedResult"/>).
/// </param>
internal record EnsureInstanceStartedResult(
EnsureInstanceStartedCommand Command, DeployedInstance? Config, string? Error,
long StoreGenerationAtRead = 0);
/// <summary>
/// Periodic instance reconcile-sweep tick (deferred register rows 78 + 72). Interval comes
/// from <see cref="SiteRuntimeOptions.InstanceSweepIntervalSeconds"/>.
/// </summary>
internal sealed record InstanceSweepTick
{
/// <summary>The singleton tick instance — the message carries no state.</summary>
public static readonly InstanceSweepTick Instance = new();
private InstanceSweepTick() { }
}
/// <summary>
/// Carries one sweep's off-thread read of the durable deployed-config rows back onto the
/// actor thread, where every comparison and repair happens.
///
/// <para>
/// <paramref name="ConnectionDefinitions"/> (rows 89 + 90) rides the same read so the DCL
/// arbitration overlay is refreshed on every tick — that is what makes a definition row
/// delivered by LocalDb CDC converge without any new cross-node message. Same null-vs-empty
/// contract as <see cref="StartupConfigsLoaded"/>: null = read failed, keep the overlay.
/// </para>
/// </summary>
/// <param name="StoreGenerationAtRead">
/// Row 101: same stale-snapshot guard as <see cref="StartupConfigsLoaded"/> — see there.
/// The sweep's copy is DROPPED on a mismatch rather than re-read: there is nothing to
/// complete, and the next tick is already a fresh read.
/// </param>
internal record InstanceSweepLoaded(
List<DeployedInstance> Configs,
string? Error,
IReadOnlyList<StoredDataConnectionDefinition>? ConnectionDefinitions = null,
long StoreGenerationAtRead = 0);
/// <summary>
/// Row 101: piped back to self once an off-thread mutation of one of the two snapshot tables
/// has COMMITTED, purely so <see cref="_storeWriteGeneration"/> can be bumped on the actor
/// thread at a point that provably happens-after the write. Used by the mutation paths that
/// have no other actor-thread completion message of their own (disable, delete, the
/// init-failure rollback); paths that already pipe a result bump in that handler instead.
/// Carries only a provenance string, for the Trace line.
/// </summary>
internal sealed record StoreWriteCompleted(string Source);
internal record DeployPersistenceResult(
string DeploymentId, string InstanceName, bool Success, string? Error,
IActorRef OriginalSender, bool IsRedeploy);
/// <summary>
/// A redeployment command buffered until the previous Instance Actor terminates.
/// </summary>
internal record PendingRedeploy(DeployInstanceCommand Command, IActorRef OriginalSender);
/// <summary>
/// Join state for a deployment awaiting confirmation. Deployment <c>Success</c> is
/// reported only once BOTH signals land: <see cref="Persisted"/> (the config committed
/// to SQLite) AND <see cref="Initialized"/> (the Instance Actor finished PreStart and
/// sent <see cref="Messages.InstanceActorInitialized"/>). 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
/// <see cref="Initialized"/>; the S6 Terminated fallback fails that deployment instead.
/// <see cref="IsRedeploy"/> gates rollback so a failed init rolls back the
/// deployed-instance set and the persisted row only for a genuinely new deployment.
/// </summary>
internal sealed class InFlightDeploy
{
/// <summary>The deployment id echoed back in the status response.</summary>
public required string DeploymentId { get; init; }
/// <summary>The deployer to answer exactly once.</summary>
public required IActorRef ReplyTo { get; init; }
/// <summary>Whether this is a redeploy (an update) rather than a new deployment.</summary>
public required bool IsRedeploy { get; init; }
/// <summary>Set when the config has committed to SQLite.</summary>
public bool Persisted { get; set; }
/// <summary>Set when the Instance Actor has finished PreStart successfully.</summary>
public bool Initialized { get; set; }
}
/// <summary>
/// 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.
/// </summary>
private sealed record RefreshFetched(RefreshDeploymentCommand Cmd, string ConfigJson, IActorRef ReplyTo);
/// <summary>
/// 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.
/// </summary>
private sealed record RefreshFetchFailed(RefreshDeploymentCommand Cmd, string Error, IActorRef ReplyTo);
/// <summary>
/// Internal message dispatched from
/// <see cref="HandleDeployArtifacts"/>'s off-thread persistence task back
/// onto the actor thread, so the DCL push (and its hash-cache mutation)
/// runs through <see cref="EnsureDclConnection"/> without crossing
/// thread-confinement boundaries. <paramref name="RemovedConnectionNames"/>
/// carries the reconcile-deleted definitions so the hash-cache eviction and
/// live DCL connection stop also happen actor-thread-confined.
/// </summary>
internal record ApplyArtifactDataConnectionsToDcl(
IReadOnlyList<Commons.Messages.Artifacts.DataConnectionArtifact> DataConnections,
IReadOnlyList<string> RemovedConnectionNames);
}
Reference in New Issue View Git Blame Copy Permalink