feat(scripting): ScriptRunGuard quarantines runaway scripts across every seam (E1)
A timed-out script is abandoned, not stopped, and every seam re-ran it on the next trigger, leaking one pool thread per trigger (R03:1.1, R05:2, R07:L1). A shared ScriptRunGuard, keyed per script instance, now: - refuses a run while that script's previous orphan is still executing (Bad "previous evaluation still running" + otopcua.script.run_refused); - quarantines after 3 consecutive non-completions or an allocation-budget breach (sampled with GC.GetAllocatedBytesForCurrentThread; default 64 MB): no further dispatch, Bad quality, one Error, otopcua.script.quarantined, cleared on source change / redeploy / reset; - starts the budget clock when the pool picks the run up, so starvation is no longer charged to the script (R07:P5), and cancels the evaluator token on timeout. Wired into RoslynVirtualTagEvaluator, CalculationEvaluator (new async EvaluateAsync; the driver's pump and timer ticks now await the run behind a SemaphoreSlim gate) and ScriptedAlarmEngine (a non-completing predicate holds state and annotates the condition BadInternalError). ScriptingOptions (Scripting:MaxAllocBytesPerRun, Scripting:QuarantineAfterConsecutiveTimeouts) carries the knobs; the Calculation driver config accepts the same two keys. Tests: guard unit suite (incl. the 30-trigger runaway leaving one orphan with a healthy sibling as positive control), per-seam refusal/quarantine/reset tests. The S12 race test now uses one vtag id per loop: twelve concurrent evaluations of ONE id is a shape production never produces and the guard refuses by design. Refs #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
@@ -73,6 +73,14 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
private readonly Func<DateTime> _clock;
|
||||
private readonly TimeSpan _scriptTimeout;
|
||||
|
||||
/// <summary>
|
||||
/// Per-alarm run quarantine (keyed by alarm id, source identity = predicate source). Refuses a
|
||||
/// predicate run while the alarm's previous (timed-out, abandoned) run is still executing and
|
||||
/// quarantines after three consecutive non-completions or an allocation-budget breach — one orphan
|
||||
/// per alarm instead of one per upstream push. Reset on every <see cref="LoadAsync"/> (a redeploy).
|
||||
/// </summary>
|
||||
private readonly ScriptRunGuard _guard;
|
||||
|
||||
// ConcurrentDictionary, not a plain Dictionary: every mutation happens under
|
||||
// _evalGate, but four read paths (GetState, GetAllStates, LoadedAlarmIds,
|
||||
// RunShelvingCheck) touch _alarms from arbitrary threads (Admin UI request
|
||||
@@ -204,13 +212,15 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
/// <param name="engineLogger">The logger for engine-level diagnostics.</param>
|
||||
/// <param name="clock">Optional function providing the current UTC time; defaults to DateTime.UtcNow.</param>
|
||||
/// <param name="scriptTimeout">Optional timeout for script execution; defaults to the evaluator's default timeout.</param>
|
||||
/// <param name="scripting">Run-guard budget + quarantine threshold; null ⇒ <see cref="ScriptingOptions"/> defaults.</param>
|
||||
public ScriptedAlarmEngine(
|
||||
ITagUpstreamSource upstream,
|
||||
IAlarmStateStore store,
|
||||
ScriptLoggerFactory loggerFactory,
|
||||
ILogger engineLogger,
|
||||
Func<DateTime>? clock = null,
|
||||
TimeSpan? scriptTimeout = null)
|
||||
TimeSpan? scriptTimeout = null,
|
||||
ScriptingOptions? scripting = null)
|
||||
{
|
||||
_upstream = upstream ?? throw new ArgumentNullException(nameof(upstream));
|
||||
_store = store ?? throw new ArgumentNullException(nameof(store));
|
||||
@@ -218,8 +228,20 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
_engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger));
|
||||
_clock = clock ?? (() => DateTime.UtcNow);
|
||||
_scriptTimeout = scriptTimeout ?? TimedScriptEvaluator<AlarmPredicateContext, bool>.DefaultTimeout;
|
||||
_guard = new ScriptRunGuard(scripting, _engineLogger);
|
||||
}
|
||||
|
||||
/// <summary>The run guard every predicate evaluation goes through (diagnostics / tests).</summary>
|
||||
public ScriptRunGuard RunGuard => _guard;
|
||||
|
||||
/// <summary>
|
||||
/// OPC UA <c>BadInternalError</c> — the quality a condition is annotated with while its predicate
|
||||
/// cannot be evaluated (timed out, refused because the previous run is still executing, or
|
||||
/// quarantined). The condition's state is frozen; the quality says why it cannot be trusted. The
|
||||
/// next completed evaluation restores the input-derived quality through the normal bucket tracking.
|
||||
/// </summary>
|
||||
private const uint StatusBadInternalError = 0x80020000u;
|
||||
|
||||
/// <summary>Raised for every emission the Part9StateMachine produces that the engine should publish.</summary>
|
||||
public event EventHandler<ScriptedAlarmEvent>? OnEvent;
|
||||
|
||||
@@ -268,6 +290,9 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
// recompile this one. Skipping this is what made the earlier fix a
|
||||
// no-op in production.
|
||||
_compileCache.Clear();
|
||||
// A (re)load is a deploy: lift every run-guard quarantine (an edited predicate already lifts
|
||||
// its own through the guard's source identity; this covers "redeployed to retry").
|
||||
_guard.ResetAll();
|
||||
|
||||
var compileFailures = new List<string>();
|
||||
foreach (var def in definitions)
|
||||
@@ -284,15 +309,17 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
|
||||
// Route through CompiledScriptCache so the emitted assembly's
|
||||
// collectible ALC participates in publish-replace cleanup.
|
||||
// The compiled script is run through _guard (timeout + quarantine) at evaluation time —
|
||||
// the per-alarm TimedScriptEvaluator this used to wrap gave the timeout but could not
|
||||
// see that the run it abandoned was still executing.
|
||||
var evaluator = _compileCache.GetOrCompile(def.PredicateScriptSource);
|
||||
var timed = new TimedScriptEvaluator<AlarmPredicateContext, bool>(evaluator, _scriptTimeout);
|
||||
var logger = _loggerFactory.Create(def.AlarmId);
|
||||
|
||||
var templateTokens = MessageTemplate.ExtractTokenPaths(def.MessageTemplate);
|
||||
var allInputs = new HashSet<string>(extraction.Reads, StringComparer.Ordinal);
|
||||
foreach (var t in templateTokens) allInputs.Add(t);
|
||||
|
||||
_alarms[def.AlarmId] = new AlarmState(def, timed, extraction.Reads, templateTokens, logger,
|
||||
_alarms[def.AlarmId] = new AlarmState(def, evaluator, extraction.Reads, templateTokens, logger,
|
||||
AlarmConditionState.Fresh(def.AlarmId, _clock()));
|
||||
|
||||
foreach (var path in allInputs)
|
||||
@@ -767,22 +794,35 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
bool predicateTrue;
|
||||
try
|
||||
{
|
||||
predicateTrue = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false);
|
||||
predicateTrue = await _guard.RunAsync(
|
||||
state.Definition.AlarmId, state.Definition.PredicateScriptSource, state.Evaluator, context,
|
||||
_scriptTimeout, ct)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (ScriptTimeoutException tex)
|
||||
{
|
||||
state.Logger.Warning("Alarm predicate timed out after {Timeout} — state unchanged", tex.Timeout);
|
||||
return seed;
|
||||
return PredicateDidNotComplete(state, seed, scratch,
|
||||
$"timed out after {tex.Timeout.TotalMilliseconds:F0} ms", pendingEmissions);
|
||||
}
|
||||
catch (ScriptRunRefusedException ex)
|
||||
{
|
||||
return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions);
|
||||
}
|
||||
catch (ScriptQuarantinedException ex)
|
||||
{
|
||||
return PredicateDidNotComplete(state, seed, scratch, ex.Message, pendingEmissions);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
scratch.LastNonCompletion = null;
|
||||
state.Logger.Error(ex, "Alarm predicate threw — state unchanged");
|
||||
return seed;
|
||||
}
|
||||
scratch.LastNonCompletion = null;
|
||||
|
||||
var result = Part9StateMachine.ApplyPredicate(seed, predicateTrue, nowUtc);
|
||||
var transition = result.Emission != EmissionKind.None
|
||||
@@ -802,6 +842,28 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
return result.State;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The predicate did not return (timed out, refused because the previous run is still executing, or
|
||||
/// quarantined): hold the condition at <paramref name="seed"/>, log the reason <b>once per change of
|
||||
/// reason</b> (a quarantined alarm is re-attempted on every push and would otherwise log every time),
|
||||
/// and annotate the condition <see cref="StatusBadInternalError"/> out of band — the frozen state
|
||||
/// cannot be trusted and the quality is how the operator sees that. The next completed evaluation
|
||||
/// restores the input-derived quality through the normal bucket tracking.
|
||||
/// </summary>
|
||||
private AlarmConditionState PredicateDidNotComplete(
|
||||
AlarmState state, AlarmConditionState seed, AlarmScratch scratch, string reason,
|
||||
List<ScriptedAlarmEvent> pendingEmissions)
|
||||
{
|
||||
if (!string.Equals(scratch.LastNonCompletion, reason, StringComparison.Ordinal))
|
||||
{
|
||||
scratch.LastNonCompletion = reason;
|
||||
state.Logger.Warning("Alarm predicate did not complete ({Reason}) — state unchanged, quality Bad", reason);
|
||||
}
|
||||
if (TrackQualityBucket(state.Definition.AlarmId, StatusBadInternalError))
|
||||
pendingEmissions.Add(BuildQualityEmission(state, seed, StatusBadInternalError));
|
||||
return seed;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Refill <paramref name="cache"/> in place from <c>_valueCache</c>, falling
|
||||
/// back to a synchronous <c>ITagUpstreamSource.ReadTag</c> for paths whose
|
||||
@@ -1201,7 +1263,7 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
|
||||
private sealed record AlarmState(
|
||||
ScriptedAlarmDefinition Definition,
|
||||
TimedScriptEvaluator<AlarmPredicateContext, bool> Evaluator,
|
||||
ScriptEvaluator<AlarmPredicateContext, bool> Evaluator,
|
||||
IReadOnlySet<string> Inputs,
|
||||
IReadOnlyList<string> TemplateTokens,
|
||||
ILogger Logger,
|
||||
@@ -1227,6 +1289,10 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
/// <summary>Gets the predicate evaluation context.</summary>
|
||||
public AlarmPredicateContext Context { get; }
|
||||
|
||||
/// <summary>The reason the last evaluation did not complete (timeout / refused / quarantined), or null
|
||||
/// after a completed run — the once-per-change log gate for <c>PredicateDidNotComplete</c>.</summary>
|
||||
public string? LastNonCompletion { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new AlarmScratch with the specified inputs, logger, and clock.
|
||||
/// </summary>
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
/// <summary>
|
||||
/// Implemented by a script seam that runs its scripts through a <see cref="ScriptRunGuard"/>, so the
|
||||
/// host that drives its deploy boundary can lift every quarantine on a redeploy (the guard's own
|
||||
/// source-change detection covers an edited script; this covers "the operator redeployed to retry").
|
||||
/// </summary>
|
||||
public interface IScriptQuarantineOwner
|
||||
{
|
||||
/// <summary>Lift every quarantine and forget every consecutive-timeout count (<see cref="ScriptRunGuard.ResetAll"/>).</summary>
|
||||
void ResetQuarantines();
|
||||
|
||||
/// <summary>The script keys currently quarantined.</summary>
|
||||
IReadOnlyCollection<string> QuarantinedScripts { get; }
|
||||
}
|
||||
@@ -0,0 +1,387 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics.Metrics;
|
||||
using Serilog;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
/// <summary>
|
||||
/// Per-script run quarantine shared by every script seam on the node. Wraps each evaluation of a
|
||||
/// <see cref="ScriptEvaluator{TContext, TResult}"/> with the wall-clock timeout
|
||||
/// <see cref="TimedScriptEvaluator{TContext, TResult}"/> used to provide, and adds what that wrapper
|
||||
/// could not: knowledge of whether the script it abandoned is <em>still running</em>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>The leak this closes.</b> A timed-out script is abandoned, not stopped — a CPU-bound
|
||||
/// <c>while(true)</c> keeps its thread-pool thread forever. Each seam re-evaluated on every trigger,
|
||||
/// so a runaway script on a 1 Hz input leaked one pool thread per second until the pool starved and
|
||||
/// unrelated work (Akka dispatchers, the OPC UA publish thread, gRPC) stalled. The guard keys every
|
||||
/// run by script instance and:
|
||||
/// </para>
|
||||
/// <list type="number">
|
||||
/// <item><b>Refuses to start a run while that script's previous run is still executing</b> —
|
||||
/// <see cref="ScriptRunRefusedException"/> (<see cref="PreviousRunStillRunningReason"/>), counted on
|
||||
/// <see cref="RefusedWhileRunningCount"/>. One orphan per script, never one per trigger.</item>
|
||||
/// <item><b>Quarantines after <see cref="ScriptingOptions.QuarantineAfterConsecutiveTimeouts"/>
|
||||
/// consecutive non-completions</b> (timeouts and still-running refusals both count; a run that
|
||||
/// returns — a value or a throw — resets the count): no further dispatch, every attempt throws
|
||||
/// <see cref="ScriptQuarantinedException"/>, one Error log, one <c>otopcua.script.quarantined</c>
|
||||
/// increment, one <see cref="Quarantined"/> event. Lifted by a source change (the operator fixed the
|
||||
/// script — a new source identity under the same key starts clean and the old orphan no longer blocks
|
||||
/// it), by <see cref="ResetAll"/> on redeploy, or by <see cref="Reset"/>.</item>
|
||||
/// <item><b>Quarantines on an allocation-budget breach</b>: the run's own-thread allocation is sampled
|
||||
/// with <see cref="GC.GetAllocatedBytesForCurrentThread"/> around the script delegate and compared
|
||||
/// with <see cref="ScriptingOptions.MaxAllocBytesPerRun"/>. This is observation after the fact — a
|
||||
/// single run that allocates enough to OOM the process is not stopped; that needs an out-of-process
|
||||
/// runner.</item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// <b>The timeout clock starts when the script starts, not when it is queued.</b> The run is pushed to
|
||||
/// the pool (a CPU-bound script would otherwise execute inline and return an already-completed task
|
||||
/// that no timeout can fire on), and the wait only begins once the pool has actually picked the run
|
||||
/// up — so thread-pool starvation is charged to nobody's script. A caller-supplied cancellation wins
|
||||
/// over the timeout (<see cref="OperationCanceledException"/>, never
|
||||
/// <see cref="ScriptTimeoutException"/>) so shutdown paths are not misread as timeouts.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Cancellation reaches the evaluator, not (yet) the script.</b> On timeout the token handed to
|
||||
/// <see cref="ScriptEvaluator{TContext, TResult}.RunAsync"/> is cancelled; a script can only observe
|
||||
/// it once <c>ScriptContext</c> exposes it (follow-up — the context type lives in the Abstractions
|
||||
/// project).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ScriptRunGuard
|
||||
{
|
||||
/// <summary>Failure reason a seam publishes when a run was refused because the previous one is still executing.</summary>
|
||||
public const string PreviousRunStillRunningReason = "previous evaluation still running";
|
||||
|
||||
/// <summary>Prefix of every quarantine reason / <see cref="ScriptQuarantinedException"/> message.</summary>
|
||||
public const string QuarantinedReasonPrefix = "script quarantined: ";
|
||||
|
||||
/// <summary>
|
||||
/// Counter <c>otopcua.script.quarantined</c> (Prometheus: <c>otopcua_script_quarantined</c>): one
|
||||
/// increment per script quarantined, tagged <c>reason</c> = <c>timeouts</c> | <c>allocation</c>.
|
||||
/// </summary>
|
||||
public static readonly Counter<long> QuarantinedCounter =
|
||||
OtOpcUaTelemetry.Meter.CreateCounter<long>("otopcua.script.quarantined", unit: "{script}",
|
||||
description: "Scripts quarantined by the run guard (reason: timeouts | allocation).");
|
||||
|
||||
/// <summary>
|
||||
/// Counter <c>otopcua.script.run_refused</c>: runs refused because the script's previous run was
|
||||
/// still executing.
|
||||
/// </summary>
|
||||
public static readonly Counter<long> RefusedCounter =
|
||||
OtOpcUaTelemetry.Meter.CreateCounter<long>("otopcua.script.run_refused", unit: "{run}",
|
||||
description: "Script runs refused because the previous run of the same script was still executing.");
|
||||
|
||||
private readonly ConcurrentDictionary<string, KeyState> _keys = new(StringComparer.Ordinal);
|
||||
private readonly long _maxAllocBytesPerRun;
|
||||
private readonly int _quarantineThreshold;
|
||||
private readonly ILogger _logger;
|
||||
private int _inFlight;
|
||||
private long _started;
|
||||
private long _refused;
|
||||
private long _quarantined;
|
||||
|
||||
/// <summary>Initializes a new guard.</summary>
|
||||
/// <param name="options">Budget + threshold; null ⇒ <see cref="ScriptingOptions"/> defaults.</param>
|
||||
/// <param name="logger">Where the quarantine Error line goes; null ⇒ Serilog's global logger.</param>
|
||||
public ScriptRunGuard(ScriptingOptions? options = null, ILogger? logger = null)
|
||||
{
|
||||
options ??= new ScriptingOptions();
|
||||
_maxAllocBytesPerRun = options.MaxAllocBytesPerRun;
|
||||
_quarantineThreshold = Math.Max(1, options.QuarantineAfterConsecutiveTimeouts);
|
||||
_logger = logger ?? Log.Logger;
|
||||
}
|
||||
|
||||
/// <summary>Raised (on the thread that detected it) when a script is quarantined.</summary>
|
||||
public event EventHandler<ScriptQuarantinedEventArgs>? Quarantined;
|
||||
|
||||
/// <summary>Runs currently executing on a pool thread — abandoned (timed-out) orphans included.</summary>
|
||||
public int InFlightCount => Volatile.Read(ref _inFlight);
|
||||
|
||||
/// <summary>Runs actually dispatched to the pool, lifetime.</summary>
|
||||
public long StartedRunCount => Interlocked.Read(ref _started);
|
||||
|
||||
/// <summary>Runs refused because the same script's previous run was still executing, lifetime.</summary>
|
||||
public long RefusedWhileRunningCount => Interlocked.Read(ref _refused);
|
||||
|
||||
/// <summary>Scripts quarantined, lifetime (a reset and re-quarantine counts twice).</summary>
|
||||
public long QuarantinedCount => Interlocked.Read(ref _quarantined);
|
||||
|
||||
/// <summary>The keys currently quarantined.</summary>
|
||||
public IReadOnlyCollection<string> QuarantinedKeys =>
|
||||
_keys.Where(kv => kv.Value.QuarantineReason is not null).Select(kv => kv.Key).ToArray();
|
||||
|
||||
/// <summary>Whether <paramref name="key"/> is quarantined, and why.</summary>
|
||||
/// <param name="key">The script instance key.</param>
|
||||
/// <param name="reason">The quarantine reason (prefixed <see cref="QuarantinedReasonPrefix"/>), or null.</param>
|
||||
/// <returns><c>true</c> when quarantined.</returns>
|
||||
public bool IsQuarantined(string key, out string? reason)
|
||||
{
|
||||
reason = _keys.TryGetValue(key, out var state) ? state.QuarantineReason : null;
|
||||
return reason is not null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Forget everything about <paramref name="key"/>: clears its quarantine and its consecutive count, and
|
||||
/// abandons its in-flight slot (an orphan still running keeps counting on <see cref="InFlightCount"/>
|
||||
/// until it finishes, but no longer blocks the key).
|
||||
/// </summary>
|
||||
/// <param name="key">The script instance key.</param>
|
||||
public void Reset(string key) => _keys.TryRemove(key, out _);
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="Reset"/> for every key — the redeploy / compile-cache-clear boundary. Orphans keep
|
||||
/// counting on <see cref="InFlightCount"/> until they finish.
|
||||
/// </summary>
|
||||
public void ResetAll() => _keys.Clear();
|
||||
|
||||
/// <summary>
|
||||
/// Run <paramref name="evaluator"/> against <paramref name="context"/> under the guard.
|
||||
/// </summary>
|
||||
/// <typeparam name="TContext">The script context type.</typeparam>
|
||||
/// <typeparam name="TResult">The script result type.</typeparam>
|
||||
/// <param name="key">The script instance key (virtual-tag id, calc-tag RawPath, alarm id).</param>
|
||||
/// <param name="sourceIdentity">
|
||||
/// The script source (or its hash). A change under the same key clears the key's quarantine and
|
||||
/// abandons its orphan slot — the operator fixed the script. Null ⇒ not tracked.
|
||||
/// </param>
|
||||
/// <param name="evaluator">The compiled script.</param>
|
||||
/// <param name="context">The script's <c>ctx</c>.</param>
|
||||
/// <param name="timeout">Wall-clock budget, measured from the moment the pool starts the run.</param>
|
||||
/// <param name="ct">Caller cancellation; wins over the timeout.</param>
|
||||
/// <returns>The script's result.</returns>
|
||||
/// <exception cref="ScriptQuarantinedException">The key is (or just became) quarantined.</exception>
|
||||
/// <exception cref="ScriptRunRefusedException">The key's previous run is still executing.</exception>
|
||||
/// <exception cref="ScriptTimeoutException">The run exceeded <paramref name="timeout"/>.</exception>
|
||||
/// <exception cref="OperationCanceledException"><paramref name="ct"/> was cancelled.</exception>
|
||||
public async Task<TResult> RunAsync<TContext, TResult>(
|
||||
string key,
|
||||
string? sourceIdentity,
|
||||
ScriptEvaluator<TContext, TResult> evaluator,
|
||||
TContext context,
|
||||
TimeSpan timeout,
|
||||
CancellationToken ct = default)
|
||||
where TContext : ScriptContext
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(key);
|
||||
ArgumentNullException.ThrowIfNull(evaluator);
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
if (timeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(timeout), "Timeout must be positive.");
|
||||
|
||||
var state = Admit(key, sourceIdentity);
|
||||
|
||||
// Slot acquisition is the whole mechanism: a second trigger landing while the previous run (orphan or
|
||||
// not) still holds the slot is refused here, before anything is dispatched to the pool.
|
||||
if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0)
|
||||
{
|
||||
Interlocked.Increment(ref _refused);
|
||||
RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
|
||||
if (NoteNonCompletion(key, state, "refused: previous evaluation still running"))
|
||||
throw new ScriptQuarantinedException(key, state.QuarantineReason!);
|
||||
throw new ScriptRunRefusedException(key);
|
||||
}
|
||||
|
||||
Interlocked.Increment(ref _inFlight);
|
||||
Interlocked.Increment(ref _started);
|
||||
|
||||
var slot = new RunSlot();
|
||||
var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var runCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
var runToken = runCts.Token;
|
||||
|
||||
// Task.Run deliberately takes NO token: a pre-cancelled token would skip the lambda, `started` would
|
||||
// never complete, and the slot would never be released. The evaluator observes the token instead.
|
||||
var runTask = Task.Run(async () =>
|
||||
{
|
||||
started.TrySetResult();
|
||||
var before = GC.GetAllocatedBytesForCurrentThread();
|
||||
try
|
||||
{
|
||||
// ScriptEvaluator.RunAsync executes the compiled delegate synchronously and returns a completed
|
||||
// task, so the after-sample is taken on the same thread, right after the script returned.
|
||||
var pending = evaluator.RunAsync(context, runToken);
|
||||
slot.AllocatedBytes = GC.GetAllocatedBytesForCurrentThread() - before;
|
||||
return await pending.ConfigureAwait(false);
|
||||
}
|
||||
catch
|
||||
{
|
||||
slot.AllocatedBytes = GC.GetAllocatedBytesForCurrentThread() - before;
|
||||
throw;
|
||||
}
|
||||
});
|
||||
|
||||
// Release the slot when the run REALLY finishes — which for an abandoned run is long after this
|
||||
// method has thrown ScriptTimeoutException. Registered before the wait so it fires regardless of
|
||||
// which way the wait below comes out. A late-finishing orphan that breached the allocation budget
|
||||
// still quarantines its key here.
|
||||
_ = runTask.ContinueWith(t =>
|
||||
{
|
||||
Interlocked.Exchange(ref state.InFlight, 0);
|
||||
Interlocked.Decrement(ref _inFlight);
|
||||
runCts.Dispose();
|
||||
if (t.IsFaulted) _ = t.Exception; // observed below by the awaiter when it is still listening
|
||||
if (t.IsCompletedSuccessfully && BreachedBudget(slot.AllocatedBytes))
|
||||
Quarantine(key, state, AllocationReason(slot.AllocatedBytes), "allocation");
|
||||
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
|
||||
|
||||
// The budget clock starts when the pool starts the run, not when we queued it (P5): pool starvation
|
||||
// is a host problem and must not be reported as "script timed out".
|
||||
await started.Task.WaitAsync(ct).ConfigureAwait(false);
|
||||
|
||||
TResult result;
|
||||
try
|
||||
{
|
||||
result = await runTask.WaitAsync(timeout, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
// The caller's cancel wins even when it lands a few µs after the timeout fired.
|
||||
if (ct.IsCancellationRequested) throw new OperationCanceledException(ct);
|
||||
runCts.Cancel();
|
||||
if (NoteNonCompletion(key, state, $"timed out after {timeout.TotalMilliseconds:F1} ms"))
|
||||
throw new ScriptQuarantinedException(key, state.QuarantineReason!);
|
||||
throw new ScriptTimeoutException(timeout);
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
// The script returned control (by throwing) — that is a completion for quarantine purposes.
|
||||
Interlocked.Exchange(ref state.ConsecutiveNonCompletions, 0);
|
||||
throw;
|
||||
}
|
||||
|
||||
Interlocked.Exchange(ref state.ConsecutiveNonCompletions, 0);
|
||||
if (BreachedBudget(slot.AllocatedBytes))
|
||||
{
|
||||
// The continuation above has already (or will have) quarantined the key; make the reason visible
|
||||
// to THIS caller too, so the breaching run itself publishes Bad rather than a suspect value.
|
||||
Quarantine(key, state, AllocationReason(slot.AllocatedBytes), "allocation");
|
||||
throw new ScriptQuarantinedException(key, state.QuarantineReason!);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>Resolve the key's state, rolling it over when the source identity changed, and refuse a quarantined key.</summary>
|
||||
private KeyState Admit(string key, string? sourceIdentity)
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
var state = _keys.GetOrAdd(key, static (_, src) => new KeyState(src), sourceIdentity);
|
||||
if (sourceIdentity is not null && state.SourceIdentity is not null
|
||||
&& !string.Equals(state.SourceIdentity, sourceIdentity, StringComparison.Ordinal))
|
||||
{
|
||||
// The script changed under this key: start from a clean slate. The old state's orphan (if any)
|
||||
// releases the GLOBAL in-flight count through its own continuation; nothing else references it.
|
||||
_keys.TryUpdate(key, new KeyState(sourceIdentity), state);
|
||||
continue;
|
||||
}
|
||||
if (sourceIdentity is not null && state.SourceIdentity is null)
|
||||
{
|
||||
state.SourceIdentity = sourceIdentity;
|
||||
}
|
||||
if (state.QuarantineReason is { } reason)
|
||||
throw new ScriptQuarantinedException(key, reason);
|
||||
return state;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Record a non-completion; returns true when it tipped the key into quarantine.</summary>
|
||||
private bool NoteNonCompletion(string key, KeyState state, string what)
|
||||
{
|
||||
var count = Interlocked.Increment(ref state.ConsecutiveNonCompletions);
|
||||
if (count < _quarantineThreshold) return false;
|
||||
Quarantine(key, state,
|
||||
$"{count} consecutive evaluation(s) did not complete (last: {what}); " +
|
||||
"the script is not evaluated again until its source changes or it is redeployed",
|
||||
"timeouts");
|
||||
return true;
|
||||
}
|
||||
|
||||
private bool BreachedBudget(long allocatedBytes) =>
|
||||
_maxAllocBytesPerRun > 0 && allocatedBytes > _maxAllocBytesPerRun;
|
||||
|
||||
private string AllocationReason(long allocatedBytes) =>
|
||||
$"one evaluation allocated {allocatedBytes / (1024.0 * 1024.0):F1} MB, over the " +
|
||||
$"{_maxAllocBytesPerRun / (1024.0 * 1024.0):F0} MB budget (Scripting:MaxAllocBytesPerRun); " +
|
||||
"the script is not evaluated again until its source changes or it is redeployed";
|
||||
|
||||
private void Quarantine(string key, KeyState state, string why, string metricReason)
|
||||
{
|
||||
var reason = QuarantinedReasonPrefix + why;
|
||||
// First writer wins: a late orphan and the awaiter can both reach here for the same run.
|
||||
if (Interlocked.CompareExchange(ref state.QuarantineReason, reason, null) is not null) return;
|
||||
|
||||
Interlocked.Increment(ref _quarantined);
|
||||
QuarantinedCounter.Add(1,
|
||||
new KeyValuePair<string, object?>("script", key),
|
||||
new KeyValuePair<string, object?>("reason", metricReason));
|
||||
_logger.Error("Script {ScriptKey} quarantined: {Reason}", key, why);
|
||||
try { Quarantined?.Invoke(this, new ScriptQuarantinedEventArgs(key, reason)); }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Warning(ex, "ScriptRunGuard Quarantined subscriber threw for {ScriptKey}", key);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class KeyState(string? sourceIdentity)
|
||||
{
|
||||
public string? SourceIdentity = sourceIdentity;
|
||||
public int InFlight;
|
||||
public int ConsecutiveNonCompletions;
|
||||
public string? QuarantineReason;
|
||||
}
|
||||
|
||||
private sealed class RunSlot
|
||||
{
|
||||
public long AllocatedBytes;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Carried by <see cref="ScriptRunGuard.Quarantined"/>.</summary>
|
||||
/// <param name="Key">The quarantined script instance key.</param>
|
||||
/// <param name="Reason">Why, prefixed <see cref="ScriptRunGuard.QuarantinedReasonPrefix"/>.</param>
|
||||
public sealed record ScriptQuarantinedEventArgs(string Key, string Reason);
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by <see cref="ScriptRunGuard.RunAsync{TContext, TResult}"/> when the script's previous run is
|
||||
/// still executing (an abandoned, timed-out run that never returned). Nothing was dispatched; the seam
|
||||
/// publishes Bad with <see cref="ScriptRunGuard.PreviousRunStillRunningReason"/>.
|
||||
/// </summary>
|
||||
public sealed class ScriptRunRefusedException : Exception
|
||||
{
|
||||
/// <summary>The script instance key.</summary>
|
||||
public string Key { get; }
|
||||
|
||||
/// <summary>Initializes the exception for <paramref name="key"/>.</summary>
|
||||
/// <param name="key">The script instance key.</param>
|
||||
public ScriptRunRefusedException(string key) : base(ScriptRunGuard.PreviousRunStillRunningReason)
|
||||
{
|
||||
Key = key;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by <see cref="ScriptRunGuard.RunAsync{TContext, TResult}"/> for a quarantined script: repeated
|
||||
/// non-completion or an allocation-budget breach. Nothing is dispatched while quarantined; the seam
|
||||
/// publishes Bad carrying the reason, which an operator clears by fixing the script (source change) or
|
||||
/// redeploying.
|
||||
/// </summary>
|
||||
public sealed class ScriptQuarantinedException : Exception
|
||||
{
|
||||
/// <summary>The script instance key.</summary>
|
||||
public string Key { get; }
|
||||
|
||||
/// <summary>Initializes the exception with the quarantine reason (already prefixed).</summary>
|
||||
/// <param name="key">The script instance key.</param>
|
||||
/// <param name="reason">The quarantine reason, prefixed <see cref="ScriptRunGuard.QuarantinedReasonPrefix"/>.</param>
|
||||
public ScriptQuarantinedException(string key, string reason) : base(reason)
|
||||
{
|
||||
Key = key;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
/// <summary>
|
||||
/// Node-wide script runtime-safety knobs (the <c>Scripting</c> appsettings section), consumed by
|
||||
/// <see cref="ScriptRunGuard"/>. One instance is shared by every script seam on the node — the
|
||||
/// virtual-tag evaluator, the Calculation evaluator and the scripted-alarm engine — so a single setting
|
||||
/// bounds every authored script the same way.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The in-process sandbox cannot pre-empt a script: a <c>while(true)</c> is abandoned on its pool
|
||||
/// thread at the timeout, and an allocation is observed only after the run returns. These knobs bound
|
||||
/// the <em>damage</em> — one orphan per script, no further runs until the source changes — not the
|
||||
/// first occurrence. Pre-emptive CPU/memory budgeting needs an out-of-process script runner.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ScriptingOptions
|
||||
{
|
||||
/// <summary>The appsettings section this binds from.</summary>
|
||||
public const string SectionName = "Scripting";
|
||||
|
||||
/// <summary>Default <see cref="MaxAllocBytesPerRun"/>: 64 MB.</summary>
|
||||
public const long DefaultMaxAllocBytesPerRun = 64L * 1024 * 1024;
|
||||
|
||||
/// <summary>Default <see cref="QuarantineAfterConsecutiveTimeouts"/>: 3.</summary>
|
||||
public const int DefaultQuarantineAfterConsecutiveTimeouts = 3;
|
||||
|
||||
/// <summary>
|
||||
/// Managed bytes one script run may allocate on its own thread (sampled with
|
||||
/// <see cref="GC.GetAllocatedBytesForCurrentThread"/> around the run) before the script is
|
||||
/// quarantined. Zero or negative disables the budget. Default 64 MB.
|
||||
/// </summary>
|
||||
public long MaxAllocBytesPerRun { get; set; } = DefaultMaxAllocBytesPerRun;
|
||||
|
||||
/// <summary>
|
||||
/// Consecutive non-completions (a run that times out, or a run refused because the previous one is
|
||||
/// still executing) after which the script is quarantined: no further runs are dispatched, every
|
||||
/// evaluation reports Bad, and the quarantine lifts only on a source change, a redeploy or an explicit
|
||||
/// reset. Default 3; values below 1 are treated as 1.
|
||||
/// </summary>
|
||||
public int QuarantineAfterConsecutiveTimeouts { get; set; } = DefaultQuarantineAfterConsecutiveTimeouts;
|
||||
}
|
||||
@@ -19,10 +19,12 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
/// task continues running on a thread-pool thread until the Roslyn runtime
|
||||
/// returns. In the CPU-bound-infinite-loop case that's effectively "leaked" —
|
||||
/// the thread is tied up until the runtime decides to return, which it may
|
||||
/// never do. Phase 7 plan Stream A.4 accepts this as a known trade-off; tighter
|
||||
/// CPU budgeting would require an out-of-process script runner, which is a v3
|
||||
/// concern. In practice, the timeout + structured warning log surfaces the
|
||||
/// offending script so the operator can fix it; the orphan thread is rare.
|
||||
/// never do. This wrapper has no memory of that orphan, so re-running the same
|
||||
/// script through it on every trigger leaks one more thread per trigger — which
|
||||
/// is why the production seams (virtual tags, Calculation, scripted alarms) run
|
||||
/// through <see cref="ScriptRunGuard"/> instead: same timeout, plus a per-script
|
||||
/// "previous run still executing" refusal and a quarantine. This class remains
|
||||
/// for callers that evaluate once (tests, the authoring preview).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Caller-supplied <see cref="CancellationToken"/> is honored — if the caller
|
||||
|
||||
Reference in New Issue
Block a user