using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.Metrics;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
///
/// Per-script run quarantine shared by every script seam on the node. Wraps each evaluation of a
/// with the wall-clock timeout
/// used to provide, and adds what that wrapper
/// could not: knowledge of whether the script it abandoned is still running.
///
///
///
/// The leak this closes. A timed-out script is abandoned, not stopped — a CPU-bound
/// while(true) 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:
///
///
/// - Refuses to start a run while that script's previous run is still executing —
/// (), counted on
/// . One orphan per script, never one per trigger.
/// - Quarantines after
/// consecutive timeouts (only a timeout is a strike; a refusal is a Bad publish plus its counter
/// and leaves the streak untouched — on a 1 Hz trigger with a 2 s budget one 4 s overrun is a
/// timeout plus two refusals and must not quarantine; a run that returns — a value or a throw —
/// resets the streak), or on a refusal behind a stuck orphan — one that has been executing for
/// longer than × the timeout (20 s at the
/// 2 s default), which is how a never-returning while(true) (one timeout, then only refusals)
/// gets quarantined: no further dispatch, every attempt throws
/// , one Error log, one otopcua.script.quarantined
/// increment, one 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 on redeploy, or by .
/// - Quarantines on an allocation-budget breach: the run's own-thread allocation is sampled
/// with around the script delegate and compared
/// with . 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.
///
///
/// The timeout clock starts when the script starts, not when it is queued. 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 (, never
/// ) so shutdown paths are not misread as timeouts.
///
///
/// Cancellation reaches the evaluator, not (yet) the script. On timeout the token handed to
/// is cancelled; a script can only observe
/// it once ScriptContext exposes it (follow-up — the context type lives in the Abstractions
/// project).
///
///
public sealed class ScriptRunGuard
{
/// Failure reason a seam publishes when a run was refused because the previous one is still executing.
public const string PreviousRunStillRunningReason = "previous evaluation still running";
/// Prefix of every quarantine reason / message.
public const string QuarantinedReasonPrefix = "script quarantined: ";
///
/// Counter otopcua.script.quarantined (Prometheus: otopcua_script_quarantined): one
/// increment per script quarantined, tagged reason = timeouts | stuck | allocation.
///
public static readonly Counter QuarantinedCounter =
OtOpcUaTelemetry.Meter.CreateCounter("otopcua.script.quarantined", unit: "{script}",
description: "Scripts quarantined by the run guard (reason: timeouts | stuck | allocation).");
///
/// Counter otopcua.script.run_refused: runs refused because the script's previous run was
/// still executing.
///
public static readonly Counter RefusedCounter =
OtOpcUaTelemetry.Meter.CreateCounter("otopcua.script.run_refused", unit: "{run}",
description: "Script runs refused because the previous run of the same script was still executing.");
private readonly ConcurrentDictionary _keys = new(StringComparer.Ordinal);
private readonly long _maxAllocBytesPerRun;
private readonly int _quarantineThreshold;
private readonly double _stuckFactor;
private readonly ILogger _logger;
private int _inFlight;
private long _started;
private long _refused;
private long _quarantined;
/// Initializes a new guard.
/// Budget + threshold; null ⇒ defaults.
/// Where the quarantine Error line goes; null ⇒ Serilog's global logger.
public ScriptRunGuard(ScriptingOptions? options = null, ILogger? logger = null)
{
// Clamped here as well as at the bind site, because options also arrive from a Calculation driver's own
// config and from tests. An unclamped factor made the refusal path's timeout * factor overflow.
options = (options ?? new ScriptingOptions()).Sanitized(out _);
_maxAllocBytesPerRun = options.MaxAllocBytesPerRun;
_quarantineThreshold = options.QuarantineAfterConsecutiveTimeouts;
_stuckFactor = options.StuckOrphanQuarantineFactor;
_logger = logger ?? Log.Logger;
}
/// Raised (on the thread that detected it) when a script is quarantined.
public event EventHandler? Quarantined;
/// Runs currently executing on a pool thread — abandoned (timed-out) orphans included.
public int InFlightCount => Volatile.Read(ref _inFlight);
/// Runs actually dispatched to the pool, lifetime.
public long StartedRunCount => Interlocked.Read(ref _started);
/// Runs refused because the same script's previous run was still executing, lifetime.
public long RefusedWhileRunningCount => Interlocked.Read(ref _refused);
/// Scripts quarantined, lifetime (a reset and re-quarantine counts twice).
public long QuarantinedCount => Interlocked.Read(ref _quarantined);
/// The keys currently quarantined.
public IReadOnlyCollection QuarantinedKeys =>
_keys.Where(kv => kv.Value.QuarantineReason is not null).Select(kv => kv.Key).ToArray();
/// Whether is quarantined, and why.
/// The script instance key.
/// The quarantine reason (prefixed ), or null.
/// true when quarantined.
public bool IsQuarantined(string key, out string? reason)
{
reason = _keys.TryGetValue(key, out var state) ? state.QuarantineReason : null;
return reason is not null;
}
///
/// Forget everything about : clears its quarantine and its consecutive count, and
/// abandons its in-flight slot (an orphan still running keeps counting on
/// until it finishes, but no longer blocks the key).
///
/// The script instance key.
public void Reset(string key) => _keys.TryRemove(key, out _);
///
/// for every key — the redeploy / compile-cache-clear boundary. Orphans keep
/// counting on until they finish.
///
public void ResetAll() => _keys.Clear();
///
/// Run against under the guard.
///
/// The script context type.
/// The script result type.
/// The script instance key (virtual-tag id, calc-tag RawPath, alarm id).
///
/// 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.
///
/// The compiled script.
/// The script's ctx.
/// Wall-clock budget, measured from the moment the pool starts the run.
/// Caller cancellation; wins over the timeout.
/// The script's result.
/// The key is (or just became) quarantined.
/// The key's previous run is still executing.
/// The run exceeded .
/// was cancelled.
public async Task RunAsync(
string key,
string? sourceIdentity,
ScriptEvaluator 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)
{
// A refusal is a Bad publish plus its counter — NOT a strike. Only a timeout (and an allocation
// breach) counts toward quarantine: on a 1 Hz trigger with a 2 s budget a single 4 s overrun is
// one timeout followed by two refusals, and counting those would quarantine a script that merely
// ran long once. The refusal also leaves the timeout streak untouched.
Interlocked.Increment(ref _refused);
RefusedCounter.Add(1, new KeyValuePair("script", key));
// …unless the orphan is STUCK: a never-returning script times out once and is then only ever
// refused, so the timeout streak alone would never quarantine it. Past the stuck threshold the
// refusal itself quarantines (same path: Error log, metric, Bad, lifted on redeploy/source change).
if (_stuckFactor > 0 && Volatile.Read(ref state.RunStartedTimestamp) is var startedAt and not 0)
{
var running = Stopwatch.GetElapsedTime(startedAt);
// Saturating: even the clamped maximum factor times a huge timeout must not overflow TimeSpan.
var thresholdTicks = timeout.Ticks * _stuckFactor;
var threshold = thresholdTicks >= TimeSpan.MaxValue.Ticks
? TimeSpan.MaxValue
: TimeSpan.FromTicks((long)thresholdTicks);
if (running > threshold)
{
Quarantine(key, state,
$"its previous evaluation has been running for {running.TotalSeconds:F1} s, over " +
$"{_stuckFactor:0.##}x its {timeout.TotalMilliseconds:F0} ms budget " +
"(Scripting:StuckOrphanQuarantineFactor) — it is stuck; the script is not evaluated again " +
"until its source changes or it is redeployed",
"stuck");
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 () =>
{
Volatile.Write(ref state.RunStartedTimestamp, Stopwatch.GetTimestamp());
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 =>
{
Volatile.Write(ref state.RunStartedTimestamp, 0);
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 (NoteTimeout(key, state, timeout))
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.ConsecutiveTimeouts, 0);
throw;
}
Interlocked.Exchange(ref state.ConsecutiveTimeouts, 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;
}
/// Resolve the key's state, rolling it over when the source identity changed, and refuse a quarantined key.
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;
}
}
/// Record a timeout strike; returns true when it tipped the key into quarantine. A completed run
/// (value or throw) resets the streak; a refusal neither counts nor resets.
private bool NoteTimeout(string key, KeyState state, TimeSpan timeout)
{
var count = Interlocked.Increment(ref state.ConsecutiveTimeouts);
if (count < _quarantineThreshold) return false;
Quarantine(key, state,
$"{count} consecutive evaluation(s) timed out (budget {timeout.TotalMilliseconds:F0} ms); " +
"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("script", key),
new KeyValuePair("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 ConsecutiveTimeouts;
/// when the pool started the in-flight run; 0 when idle.
public long RunStartedTimestamp;
public string? QuarantineReason;
}
private sealed class RunSlot
{
public long AllocatedBytes;
}
}
/// Carried by .
/// The quarantined script instance key.
/// Why, prefixed .
public sealed record ScriptQuarantinedEventArgs(string Key, string Reason);
///
/// Thrown by 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 .
///
public sealed class ScriptRunRefusedException : Exception
{
/// The script instance key.
public string Key { get; }
/// Initializes the exception for .
/// The script instance key.
public ScriptRunRefusedException(string key) : base(ScriptRunGuard.PreviousRunStillRunningReason)
{
Key = key;
}
}
///
/// Thrown by for a quarantined script: repeated
/// timeouts, a stuck orphan 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.
///
public sealed class ScriptQuarantinedException : Exception
{
/// The script instance key.
public string Key { get; }
/// Initializes the exception with the quarantine reason (already prefixed).
/// The script instance key.
/// The quarantine reason, prefixed .
public ScriptQuarantinedException(string key, string reason) : base(reason)
{
Key = key;
}
}