dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(runtime): split trigger evals from the blocking pool; bounded, deadline-aware execution

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-14 22:36:15 -04:00
parent 312216ff2b
commit c4fc1f8ecd
42 changed files with 3673 additions and 1251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs
+311 -28
View File
@@ -142,6 +142,23 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// </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>Akka timer scheduler injected by the framework via <see cref="IWithTimers"/>.</summary>
public ITimerScheduler Timers { get; set; } = null!;
@@ -172,6 +189,11 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// 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>
public DeploymentManagerActor(
SiteStorageService storage,
ScriptCompilationService compilationService,
@@ -186,8 +208,10 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
IDeploymentConfigFetcher? configFetcher = null,
TimeSpan? startupLoadRetryInterval = null,
Func<Task<List<DeployedInstance>>>? configLoader = null,
ExternalSystemDefinitionCache? externalSystemCache = null)
ExternalSystemDefinitionCache? externalSystemCache = null,
ScriptExecutionScheduler? scriptScheduler = null)
{
_scriptScheduler = scriptScheduler;
_storage = storage;
_compilationService = compilationService;
_deployCompileValidator = new DeployCompileValidator(compilationService);
@@ -212,9 +236,16 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// Lifecycle commands
Receive<DeployInstanceCommand>(cmd => HandleDeploy(cmd, Sender));
Receive<DisableInstanceCommand>(HandleDisable);
Receive<EnableInstanceCommand>(HandleEnable);
Receive<DeleteInstanceCommand>(HandleDelete);
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
@@ -461,8 +492,15 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
}
/// <summary>
/// Creates Instance Actors in batches with a configurable delay between batches
/// to prevent reconnection storms on failover.
/// 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)
{
@@ -471,6 +509,45 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
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 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);
@@ -520,23 +597,164 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// apply still replies to the right actor.
/// </param>
private void HandleDeploy(DeployInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
// 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
/// process-wide <see cref="SiteScriptCompileCache"/>, 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 runs
// synchronously on the actor thread: it is a pure prefix step of the
// deploy handler, so the existing redeploy-supersede / delete-during-
// redeploy ordering (which depends on strict mailbox FIFO) is preserved
// exactly. It is NOT run off-thread — piping the verdict back to self
// reorders concurrent deploys relative to each other and to
// delete/disable commands, breaking that ordering. A deploy is an
// infrequent admin command, so briefly holding the singleton for a pure
// Roslyn compile is acceptable; the central deployer already Asks and
// waits for the DeploymentStatusResponse. Redeploys and multi-instance
// deploys of unchanged scripts hit the process-wide compile cache
// (SiteScriptCompileCache), so the synchronous gate recompiles only
// genuinely new code and the Instance Actor's PreStart reuses the gate's
// compile (N4).
// 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 process-wide compile
// cache (SiteScriptCompileCache) too, and the Instance Actor's PreStart reuses
// the gate's compile (N4).
var errors = _deployCompileValidator.Validate(command.FlattenedConfigurationJson);
if (errors.Count > 0)
{
@@ -942,10 +1160,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// <summary>
/// Disables an instance: stops the actor and marks as disabled in SQLite.
/// </summary>
private void HandleDisable(DisableInstanceCommand command)
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;
// A disable arriving mid-redeploy must cancel the buffered
// redeploy. Otherwise HandleTerminated re-creates the Instance Actor and
// re-stores its config with isEnabled: true when the predecessor terminates,
@@ -975,7 +1197,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
UpdateInstanceCounts();
var sender = Sender;
var sender = replyTo;
_storage.SetInstanceEnabledAsync(instanceName, false).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
@@ -1005,10 +1227,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// Enables an instance: marks as enabled in SQLite and re-creates the Instance Actor
/// from the stored config.
/// </summary>
private void HandleEnable(EnableInstanceCommand command)
private void HandleEnable(EnableInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
var sender = Sender;
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
var sender = replyTo;
Task.Run(async () =>
{
@@ -1061,10 +1287,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// 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)
private void HandleDelete(DeleteInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
// A delete arriving while a redeploy is still terminating must
// be authoritative over the mid-redeploy bookkeeping. HandleDeploy already
// removed the instance from _instanceActors and buffered a PendingRedeploy
@@ -1105,7 +1334,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
_deployedInstanceNames.Remove(instanceName);
UpdateInstanceCounts();
var sender = Sender;
var sender = replyTo;
_storage.RemoveDeployedConfigAsync(instanceName).ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
@@ -1135,7 +1364,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// 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/ScriptExecutionActor
/// deployment pipeline (matching the established ScriptActor / script-run
/// pattern).
/// <para>
/// <b>Thread-safety:</b> the disable (<see cref="HandleDisable"/>) and delete
@@ -2089,7 +2318,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
internal int InstanceActorCount => _instanceActors.Count;
/// <summary>
/// Updates the health collector with current instance counts.
/// 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()
@@ -2098,6 +2328,22 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
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 ──
@@ -2131,6 +2377,43 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
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(