From eb9269c337d5ce074deb075b83bec1b54351b438 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:39:43 -0400 Subject: [PATCH] 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 --- .../ScriptedAlarmEngine.cs | 82 +++- .../IScriptQuarantineOwner.cs | 15 + .../ScriptRunGuard.cs | 387 ++++++++++++++++++ .../ScriptingOptions.cs | 42 ++ .../TimedScriptEvaluator.cs | 10 +- .../CalculationDriver.cs | 83 ++-- .../CalculationDriverFactoryExtensions.cs | 17 + .../CalculationDriverOptions.cs | 21 +- .../CalculationEvaluator.cs | 75 +++- .../CalculationTimerScheduler.cs | 20 +- .../Engines/RoslynVirtualTagEvaluator.cs | 56 ++- .../VirtualTags/VirtualTagHostActor.cs | 6 + .../ScriptedAlarmEngineRunGuardTests.cs | 105 +++++ .../ScriptRunGuardTests.cs | 306 ++++++++++++++ .../CalculationEvaluatorRunGuardTests.cs | 86 ++++ .../RoslynVirtualTagEvaluatorRunGuardTests.cs | 66 +++ .../RoslynVirtualTagEvaluatorTests.cs | 10 +- 17 files changed, 1313 insertions(+), 74 deletions(-) create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptQuarantineOwner.cs create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs index 4ccc3592..808935ae 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs @@ -73,6 +73,14 @@ public sealed class ScriptedAlarmEngine : IDisposable private readonly Func _clock; private readonly TimeSpan _scriptTimeout; + /// + /// 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 (a redeploy). + /// + 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 /// The logger for engine-level diagnostics. /// Optional function providing the current UTC time; defaults to DateTime.UtcNow. /// Optional timeout for script execution; defaults to the evaluator's default timeout. + /// Run-guard budget + quarantine threshold; null ⇒ defaults. public ScriptedAlarmEngine( ITagUpstreamSource upstream, IAlarmStateStore store, ScriptLoggerFactory loggerFactory, ILogger engineLogger, Func? 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.DefaultTimeout; + _guard = new ScriptRunGuard(scripting, _engineLogger); } + /// The run guard every predicate evaluation goes through (diagnostics / tests). + public ScriptRunGuard RunGuard => _guard; + + /// + /// OPC UA BadInternalError — 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. + /// + private const uint StatusBadInternalError = 0x80020000u; + /// Raised for every emission the Part9StateMachine produces that the engine should publish. public event EventHandler? 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(); 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(evaluator, _scriptTimeout); var logger = _loggerFactory.Create(def.AlarmId); var templateTokens = MessageTemplate.ExtractTokenPaths(def.MessageTemplate); var allInputs = new HashSet(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; } + /// + /// The predicate did not return (timed out, refused because the previous run is still executing, or + /// quarantined): hold the condition at , log the reason once per change of + /// reason (a quarantined alarm is re-attempted on every push and would otherwise log every time), + /// and annotate the condition 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. + /// + private AlarmConditionState PredicateDidNotComplete( + AlarmState state, AlarmConditionState seed, AlarmScratch scratch, string reason, + List 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; + } + /// /// Refill in place from _valueCache, falling /// back to a synchronous ITagUpstreamSource.ReadTag for paths whose @@ -1201,7 +1263,7 @@ public sealed class ScriptedAlarmEngine : IDisposable private sealed record AlarmState( ScriptedAlarmDefinition Definition, - TimedScriptEvaluator Evaluator, + ScriptEvaluator Evaluator, IReadOnlySet Inputs, IReadOnlyList TemplateTokens, ILogger Logger, @@ -1227,6 +1289,10 @@ public sealed class ScriptedAlarmEngine : IDisposable /// Gets the predicate evaluation context. public AlarmPredicateContext Context { get; } + /// The reason the last evaluation did not complete (timeout / refused / quarantined), or null + /// after a completed run — the once-per-change log gate for PredicateDidNotComplete. + public string? LastNonCompletion { get; set; } + /// /// Initializes a new AlarmScratch with the specified inputs, logger, and clock. /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptQuarantineOwner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptQuarantineOwner.cs new file mode 100644 index 00000000..3cf38fac --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/IScriptQuarantineOwner.cs @@ -0,0 +1,15 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; + +/// +/// Implemented by a script seam that runs its scripts through a , 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"). +/// +public interface IScriptQuarantineOwner +{ + /// Lift every quarantine and forget every consecutive-timeout count (). + void ResetQuarantines(); + + /// The script keys currently quarantined. + IReadOnlyCollection QuarantinedScripts { get; } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs new file mode 100644 index 00000000..8f1acefd --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs @@ -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; + +/// +/// 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 non-completions (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 +/// , 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 | allocation. + /// + public static readonly Counter QuarantinedCounter = + OtOpcUaTelemetry.Meter.CreateCounter("otopcua.script.quarantined", unit: "{script}", + description: "Scripts quarantined by the run guard (reason: timeouts | 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 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) + { + options ??= new ScriptingOptions(); + _maxAllocBytesPerRun = options.MaxAllocBytesPerRun; + _quarantineThreshold = Math.Max(1, options.QuarantineAfterConsecutiveTimeouts); + _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) + { + Interlocked.Increment(ref _refused); + RefusedCounter.Add(1, new KeyValuePair("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; + } + + /// 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 non-completion; returns true when it tipped the key into quarantine. + 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("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 ConsecutiveNonCompletions; + 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 +/// 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. +/// +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; + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs new file mode 100644 index 00000000..ba1399fa --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs @@ -0,0 +1,42 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; + +/// +/// Node-wide script runtime-safety knobs (the Scripting appsettings section), consumed by +/// . 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. +/// +/// +/// +/// The in-process sandbox cannot pre-empt a script: a while(true) is abandoned on its pool +/// thread at the timeout, and an allocation is observed only after the run returns. These knobs bound +/// the damage — 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. +/// +/// +public sealed class ScriptingOptions +{ + /// The appsettings section this binds from. + public const string SectionName = "Scripting"; + + /// Default : 64 MB. + public const long DefaultMaxAllocBytesPerRun = 64L * 1024 * 1024; + + /// Default : 3. + public const int DefaultQuarantineAfterConsecutiveTimeouts = 3; + + /// + /// Managed bytes one script run may allocate on its own thread (sampled with + /// around the run) before the script is + /// quarantined. Zero or negative disables the budget. Default 64 MB. + /// + public long MaxAllocBytesPerRun { get; set; } = DefaultMaxAllocBytesPerRun; + + /// + /// 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. + /// + public int QuarantineAfterConsecutiveTimeouts { get; set; } = DefaultQuarantineAfterConsecutiveTimeouts; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/TimedScriptEvaluator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/TimedScriptEvaluator.cs index 4adeccab..7a860436 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/TimedScriptEvaluator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/TimedScriptEvaluator.cs @@ -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 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). /// /// /// Caller-supplied is honored — if the caller diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs index d556ccf6..dc0d8ba1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs @@ -92,7 +92,15 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba // the timer ticks — the mutual exclusion _lock used to provide while it was held across the run, minus // the blocking. Taken ONLY by evaluation workers, so no host thread (mux callback, ReadAsync, // reinit) can ever wait on a script. Lock order is always _evalGate → _lock, never the reverse. - private readonly object _evalGate = new(); + // A SemaphoreSlim rather than a monitor because the holder AWAITS the script run (the pump is async — + // it no longer parks a second pool thread behind the one the script runs on, and the run-guard's + // budget clock starts only once the pool has picked the script up, so pool starvation is never + // reported as a script timeout). + private readonly SemaphoreSlim _evalGate = new(1, 1); + + // Cancels an awaited script run on Dispose (the only hard teardown — ShutdownAsync leaves an in-flight + // run to its RunTimeout, exactly as before, since the pump drops its result once _running is false). + private readonly CancellationTokenSource _evalCts = new(); // Change-triggered evaluations queued by OnDependencyValue and drained, in arrival order, by a single // pump task. Both fields are guarded by _lock; _pumpScheduled stays true until the pump observes an @@ -129,7 +137,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba // _logger (ILogger) implements the non-generic ILogger, so it is passed straight // through — the evaluator's host-side diagnostics (compile errors, sandbox violations, execution // throws) land on the SAME logger the driver itself was constructed with, not a separate one. - _evaluator = new CalculationEvaluator(_logger, _scriptRoot, _options.RunTimeout); + _evaluator = new CalculationEvaluator(_logger, _scriptRoot, _options.RunTimeout, _options.Scripting); RebuildTagTable(_options); } @@ -225,6 +233,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba { RawTags = dto.RawTags is { Count: > 0 } ? [.. dto.RawTags] : [], RunTimeout = _options.RunTimeout, + Scripting = _options.Scripting, }; } } @@ -235,7 +244,10 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba _driverInstanceId); } } - return new CalculationDriverOptions { RawTags = _options.RawTags, RunTimeout = _options.RunTimeout }; + return new CalculationDriverOptions + { + RawTags = _options.RawTags, RunTimeout = _options.RunTimeout, Scripting = _options.Scripting, + }; } /// Re-bind the tag table from — shared by @@ -491,13 +503,15 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba { if (_pumpScheduled) return; _pumpScheduled = true; - _ = Task.Run(RunEvaluationPump); + _ = Task.Run(RunEvaluationPumpAsync); } - /// Drain the queued change-triggered evaluations, in arrival order, on a pool thread. At most - /// one pump runs at a time, so per-tag evaluation ordering is preserved exactly as it was when the - /// change path evaluated inline under the state lock. - private void RunEvaluationPump() + /// Drain the queued change-triggered evaluations, in arrival order. At most one pump runs at a + /// time, so per-tag evaluation ordering is preserved exactly as it was when the change path evaluated + /// inline under the state lock. Async: the pump AWAITS each script run instead of parking a pool thread + /// behind the one the script executes on (P5 — sync-over-async with a nested Task.Run manufactured + /// false timeouts under pool starvation). + private async Task RunEvaluationPumpAsync() { while (true) { @@ -515,39 +529,50 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba // _evalGate (never _lock) is held across the run, so the pump and the timer ticks still // evaluate one at a time while no host thread can block behind a script. - lock (_evalGate) + await _evalGate.WaitAsync().ConfigureAwait(false); + try { - try - { - RunEvaluation(next); - } - catch (Exception ex) - { - // The evaluator never throws for script faults, so this is a defect path only — it - // must not kill the pump and strand every queued evaluation behind it. - _logger.LogError(ex, "Calculation {Driver}: evaluation of {RawPath} failed", - _driverInstanceId, next.Definition.RawPath); - } + await RunEvaluationAsync(next).ConfigureAwait(false); + } + catch (Exception ex) + { + // The evaluator never throws for script faults, so this is a defect path only — it + // must not kill the pump and strand every queued evaluation behind it. + _logger.LogError(ex, "Calculation {Driver}: evaluation of {RawPath} failed", + _driverInstanceId, next.Definition.RawPath); + } + finally + { + _evalGate.Release(); } } } - /// Timer-trigger entry point — prepares and evaluates a single calc tag on the scheduler's - /// pool thread (the scheduler's per-group in-flight guard is what bounds outstanding timer work). - private void EvaluateTagFromTimer(string rawPath) + /// Timer-trigger entry point — prepares and evaluates a single calc tag (the scheduler's + /// per-group in-flight guard is what bounds outstanding timer work; it stays set until this task + /// completes). + private async Task EvaluateTagFromTimerAsync(string rawPath) { if (!_running) return; PendingEvaluation? prepared; lock (_lock) prepared = PrepareEvaluation(rawPath, DateTime.UtcNow); if (prepared is null) return; - lock (_evalGate) RunEvaluation(prepared); + await _evalGate.WaitAsync().ConfigureAwait(false); + try + { + await RunEvaluationAsync(prepared).ConfigureAwait(false); + } + finally + { + _evalGate.Release(); + } } /// Run one prepared evaluation and publish the outcome. MUST be called under /// and NOT under : the script run is the whole point of the /// split, and the script log + fan-out both invoke subscriber code. /// The prepared evaluation. - private void RunEvaluation(PendingEvaluation pending) + private async Task RunEvaluationAsync(PendingEvaluation pending) { var def = pending.Definition; @@ -572,7 +597,8 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba // Snapshots, not bare values: the script's ctx.GetTag(path).StatusCode / .SourceTimestampUtc are // then the input's real ones rather than a fabricated Good stamped at evaluation time. - var result = _evaluator.Evaluate(def.RawPath, def.ScriptSource, pending.Dependencies); + var result = await _evaluator.EvaluateAsync(def.RawPath, def.ScriptSource, pending.Dependencies, _evalCts.Token) + .ConfigureAwait(false); string? failureReason = null; (object? Value, uint StatusCode)? toPublish; @@ -688,7 +714,7 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba { var timed = _tagsByRawPath.Values.Where(t => t.TimerInterval is not null).ToArray(); if (timed.Length == 0) return; - _timers = new CalculationTimerScheduler(EvaluateTagFromTimer, _logger); + _timers = new CalculationTimerScheduler(EvaluateTagFromTimerAsync, _logger); _timers.Start(timed); } @@ -723,6 +749,9 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba _running = false; StopTimers(); lock (_lock) _pending.Clear(); + // Cancel an awaited run first (the guard lets the caller's cancel win over the timeout), then drop + // the compiled scripts. The CTS is not disposed: a pump mid-await may still read its token. + try { _evalCts.Cancel(); } catch (ObjectDisposedException) { } _evaluator.Dispose(); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs index ab10f06d..349a0cf1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs @@ -71,11 +71,22 @@ public static class CalculationDriverFactoryExtensions ?? throw new InvalidOperationException( $"Calculation driver config for '{driverInstanceId}' deserialised to null"); + // Run-guard knobs are optional and default to the node-wide ScriptingOptions defaults; only an + // explicitly authored key overrides one (so a config carrying neither binds Scripting = null). + ScriptingOptions? scripting = null; + if (dto.MaxAllocBytesPerRun is not null || dto.QuarantineAfterConsecutiveTimeouts is not null) + { + scripting = new ScriptingOptions(); + if (dto.MaxAllocBytesPerRun is { } alloc) scripting.MaxAllocBytesPerRun = alloc; + if (dto.QuarantineAfterConsecutiveTimeouts is { } n) scripting.QuarantineAfterConsecutiveTimeouts = n; + } + var options = new CalculationDriverOptions { RawTags = dto.RawTags is { Count: > 0 } ? [.. dto.RawTags] : [], RunTimeout = dto.RunTimeoutMs is { } ms && ms > 0 ? TimeSpan.FromMilliseconds(ms) : TimeSpan.FromSeconds(2), + Scripting = scripting, }; return new CalculationDriver( @@ -92,5 +103,11 @@ public static class CalculationDriverFactoryExtensions /// Optional per-script evaluation wall-clock budget override, in milliseconds (default 2000). public int? RunTimeoutMs { get; init; } + + /// Optional run-guard allocation budget per script run, in bytes (default 64 MB; ≤ 0 disables). + public long? MaxAllocBytesPerRun { get; init; } + + /// Optional run-guard quarantine threshold: consecutive non-completions before a tag's script stops being evaluated (default 3). + public int? QuarantineAfterConsecutiveTimeouts { get; init; } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverOptions.cs index b6642c09..f00f50e4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverOptions.cs @@ -1,20 +1,21 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation; -/// -/// Bound options for a instance. The Calculation pseudo-driver has -/// no backend/endpoint config — its only input is the authored raw calc tags the deploy artifact -/// injects as the list (each carrying the tag's scriptId + resolved -/// scriptSource + trigger config inside its blob). -/// +/// Bound options for one instance. public sealed class CalculationDriverOptions { - /// The driver's authored raw calc tags (RawPath + TagConfig blob + WriteIdempotent), from - /// the deploy artifact. Each is mapped to a at construction. + /// The authored raw calc tags (RawPath + TagConfig JSON) this driver computes. public IReadOnlyList RawTags { get; init; } = Array.Empty(); - /// Per-script wall-clock evaluation budget handed to the . - /// Defaults to 2 seconds (parity with the VirtualTag evaluator). + /// Wall-clock budget per script run (driver-config runTimeoutMs); default 2 s. public TimeSpan RunTimeout { get; init; } = TimeSpan.FromSeconds(2); + + /// + /// Run-guard budget + quarantine threshold for this driver's scripts (driver-config + /// maxAllocBytesPerRun / quarantineAfterConsecutiveTimeouts). Null ⇒ the + /// defaults (64 MB, 3). + /// + public ScriptingOptions? Scripting { get; init; } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs index 249c6958..4296b993 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs @@ -37,8 +37,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation; /// apply-boundary clear, which is what stops collectible AssemblyLoadContexts accreting across /// script edits. /// +/// +/// Run quarantine. Every compiled run goes through one keyed by +/// calc-tag RawPath with the script source as the identity: a run is refused while the tag's previous +/// (timed-out, abandoned) run is still executing, and the tag is quarantined after three consecutive +/// non-completions or an allocation-budget breach — both surface as a +/// the driver maps to Bad. The driver's evaluation pump awaits rather than +/// blocking a pool thread on the sync , +/// and the guard starts the budget clock only once the pool has picked the script up, so pool starvation +/// is never reported as a script timeout. +/// /// -public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable +public sealed class CalculationEvaluator : IScriptCacheOwner, IScriptQuarantineOwner, IDisposable { // Source-hash-keyed compile cache: Lazy single-compile (no double-compile race) and Clear()/Dispose() // that dispose each evaluator's collectible AssemblyLoadContext. ClearCompiledScripts() — driven by the @@ -47,6 +57,7 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable private readonly ILogger _logger; private readonly SerilogLogger _scriptRoot; private readonly TimeSpan _runTimeout; + private readonly ScriptRunGuard _guard; private bool _disposed; /// Initializes a new . @@ -56,16 +67,22 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable /// implements this — instead of needing a second, separately-obtained logger instance. /// Root script logger; user ctx.Logger.* output flows through this to the Script-log page. /// Wall-clock budget per script run; defaults to 2 seconds (parity with the VT evaluator). + /// Run-guard budget + quarantine threshold; null ⇒ defaults. public CalculationEvaluator( ILogger logger, ScriptRootLogger scriptRoot, - TimeSpan? runTimeout = null) + TimeSpan? runTimeout = null, + ScriptingOptions? scripting = null) { _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _scriptRoot = (scriptRoot ?? throw new ArgumentNullException(nameof(scriptRoot))).Logger; _runTimeout = runTimeout ?? TimeSpan.FromSeconds(2); + _guard = new ScriptRunGuard(scripting, _scriptRoot); } + /// The run guard every compiled evaluation goes through (diagnostics / tests). + public ScriptRunGuard RunGuard => _guard; + /// /// Evaluate for the calc tag against /// the snapshot in (RawPath → last value). Never throws — script @@ -112,8 +129,25 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable /// The C# script source producing the tag's value. /// Read-only snapshot of dependency values keyed by RawPath. /// The evaluation outcome. + /// Blocks the calling thread for the run. The driver's pump uses ; + /// this overload remains for callers with no async context — chiefly tests. public VirtualTagEvalResult Evaluate( string calcTagId, string scriptSource, IReadOnlyDictionary dependencies) + => EvaluateAsync(calcTagId, scriptSource, dependencies).GetAwaiter().GetResult(); + + /// + /// The async form of : + /// the compiled script runs on the pool under the and the caller awaits + /// it instead of blocking a second pool thread for the duration. Never throws for script faults. + /// + /// Identity of the calc tag (bound to the script log). + /// The C# script source producing the tag's value. + /// Read-only snapshot of dependency values keyed by RawPath. + /// Caller cancellation (driver shutdown); wins over the run timeout. + /// The evaluation outcome. + public async Task EvaluateAsync( + string calcTagId, string scriptSource, IReadOnlyDictionary dependencies, + CancellationToken ct = default) { if (_disposed) return VirtualTagEvalResult.Failure("evaluator disposed"); if (string.IsNullOrWhiteSpace(scriptSource)) return VirtualTagEvalResult.Failure("empty expression"); @@ -180,11 +214,12 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable try { - // Route through TimedScriptEvaluator (Task.Run + WaitAsync): a raw CancellationToken can't - // interrupt a CPU-bound/infinite-loop Roslyn script, so the timed wrapper is what bounds a - // runaway script to the wall-clock budget instead of wedging the driver's dispatch. - var timed = new TimedScriptEvaluator(evaluator, _runTimeout); - var raw = timed.RunAsync(context).GetAwaiter().GetResult(); + // Route through the ScriptRunGuard (Task.Run + WaitAsync, keyed by RawPath): a raw + // CancellationToken can't interrupt a CPU-bound/infinite-loop Roslyn script, so the wall-clock + // budget is what bounds a runaway script — and the guard is what keeps the next trigger from + // stacking another orphan behind it (one per tag, then quarantine). + var raw = await _guard.RunAsync(calcTagId, scriptSource, evaluator, context, _runTimeout, ct) + .ConfigureAwait(false); return VirtualTagEvalResult.Ok(raw); } catch (ScriptTimeoutException) @@ -192,6 +227,18 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable return VirtualTagEvalResult.Failure( $"{ScriptEvalFailureReasons.TimedOutPrefix}{_runTimeout.TotalSeconds:F1}s"); } + catch (ScriptRunRefusedException ex) + { + return VirtualTagEvalResult.Failure(ex.Message); + } + catch (ScriptQuarantinedException ex) + { + return VirtualTagEvalResult.Failure(ex.Message); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return VirtualTagEvalResult.Failure("evaluation cancelled (driver shutting down)"); + } catch (ObjectDisposedException) when (!_disposed && attempt == 0) { // A concurrent ClearCompiledScripts disposed the evaluator between our GetOrCompile and the @@ -212,7 +259,19 @@ public sealed class CalculationEvaluator : IScriptCacheOwner, IDisposable } /// - public void ClearCompiledScripts() => _cache.Clear(); + public void ClearCompiledScripts() + { + _cache.Clear(); + // The compile-cache clear is this evaluator's deploy boundary — a redeploy lifts every quarantine + // (an edited source already lifts its own through the guard's source identity). + _guard.ResetAll(); + } + + /// + public void ResetQuarantines() => _guard.ResetAll(); + + /// + public IReadOnlyCollection QuarantinedScripts => _guard.QuarantinedKeys; /// Disposes the evaluator, draining + disposing every cached compile's collectible ALC. public void Dispose() diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs index ef74da0e..694f7d8a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationTimerScheduler.cs @@ -7,11 +7,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation; /// per distinct interval group (so the wire count stays low /// regardless of tag count) with a per-group in-flight flag that skips a tick when the prior tick for /// the same group is still running. Each tick invokes the driver's own evaluate -/// callback per tag in the group. +/// callback per tag in the group and AWAITS it — the in-flight flag is held across the awaited script +/// run, not just across the synchronous part of the callback, so a slow script still bounds its group +/// to one outstanding tick. /// internal sealed class CalculationTimerScheduler : IDisposable { - private readonly Action _evaluate; + private readonly Func _evaluate; private readonly ILogger _logger; private readonly List _timers = []; private readonly List _groups = []; @@ -20,9 +22,9 @@ internal sealed class CalculationTimerScheduler : IDisposable private bool _disposed; /// Initializes a new instance of the class. - /// Callback invoked with a calc tag's RawPath when its timer group ticks. + /// Callback invoked with a calc tag's RawPath when its timer group ticks; awaited. /// Logger for diagnostics. - public CalculationTimerScheduler(Action evaluate, ILogger logger) + public CalculationTimerScheduler(Func evaluate, ILogger logger) { _evaluate = evaluate ?? throw new ArgumentNullException(nameof(evaluate)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -69,12 +71,20 @@ internal sealed class CalculationTimerScheduler : IDisposable return; } + // The Timer callback is void; the tick itself is async so the awaited script runs do not pin this + // pool thread. TickAsync owns the in-flight flag and swallows everything — nothing can escape to the + // Timer (which would be an unobserved fault on a timer thread). + _ = TickAsync(ctx); + } + + private async Task TickAsync(TickGroup ctx) + { try { foreach (var path in ctx.Paths) { if (_cts.IsCancellationRequested) return; - try { _evaluate(path); } + try { await _evaluate(path).ConfigureAwait(false); } catch (Exception ex) { _logger.LogError(ex, "CalculationTimerScheduler evaluate failed for {Path}", path); } } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs index d77631ea..c6823c41 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs @@ -31,8 +31,17 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines; /// into the fresh cache for one generation; P7's expression-set gate means no-op redeploys no /// longer open this window at all. /// +/// +/// Run quarantine. Every compiled run goes through one keyed by +/// virtual-tag id with the expression as the source identity: a run is refused while the tag's previous +/// (timed-out, abandoned) run is still executing, and the tag is quarantined after three consecutive +/// non-completions or an allocation-budget breach. Both surface as +/// so VirtualTagActor publishes Bad exactly as it does for a timeout. A quarantine lifts when the +/// expression changes, on / (the deploy +/// boundary), never by itself. +/// /// -public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCacheOwner, IDisposable +public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCacheOwner, IScriptQuarantineOwner, IDisposable { // Purpose-built source-hash-keyed compile cache: Lazy single-compile (no double-compile race) and // Clear()/Dispose() that dispose each evaluator's collectible AssemblyLoadContext. The apply-boundary @@ -42,22 +51,29 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac private readonly ILogger _logger; private readonly SerilogLogger _scriptRoot; private readonly TimeSpan _runTimeout; + private readonly ScriptRunGuard _guard; private bool _disposed; /// Initializes a new RoslynVirtualTagEvaluator with the given loggers and optional timeout. /// Logger for recording compilation and execution errors (host diagnostics). /// Root script logger; user ctx.Logger.* output flows through this to the Script-log page. /// Maximum execution time for each script; defaults to 2 seconds if not specified. + /// Run-guard budget + quarantine threshold (the Scripting section); null ⇒ defaults. public RoslynVirtualTagEvaluator( ILogger logger, ScriptRootLogger scriptRoot, - TimeSpan? runTimeout = null) + TimeSpan? runTimeout = null, + ScriptingOptions? scripting = null) { _logger = logger; _scriptRoot = (scriptRoot ?? throw new ArgumentNullException(nameof(scriptRoot))).Logger; _runTimeout = runTimeout ?? TimeSpan.FromSeconds(2); + _guard = new ScriptRunGuard(scripting, _scriptRoot); } + /// The run guard every compiled evaluation goes through (diagnostics / tests). + public ScriptRunGuard RunGuard => _guard; + // Value-only overload. Every dependency reaches the script stamped Good with an // evaluation-time timestamp — bare values carry nothing better. The runtime calls the snapshot // overload instead; this one remains for callers that genuinely hold only @@ -150,13 +166,13 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac try { - // Route through TimedScriptEvaluator (Task.Run + WaitAsync): a raw CancellationToken can't - // interrupt a CPU-bound/infinite-loop script (Roslyn scripts don't poll it), so the previous - // CTS-only path let one runaway script hang this actor forever. WaitAsync returns control when - // the wall-clock budget fires regardless of whether the inner task completes (the orphaned - // thread is the documented, accepted trade-off — TimedScriptEvaluator remarks). - var timed = new TimedScriptEvaluator(evaluator, _runTimeout); - var raw = timed.RunAsync(context).GetAwaiter().GetResult(); + // Route through the ScriptRunGuard (Task.Run + WaitAsync, keyed by virtual-tag id): a raw + // CancellationToken can't interrupt a CPU-bound/infinite-loop script (Roslyn scripts don't poll + // it), so the wall-clock budget is what returns control — and the guard is what stops the + // abandoned run from being joined by another on the next trigger (one orphan per tag, then + // quarantine), which is the leak the old per-evaluation TimedScriptEvaluator allowed. + var raw = _guard.RunAsync(virtualTagId, expression, evaluator, context, _runTimeout) + .GetAwaiter().GetResult(); return VirtualTagEvalResult.Ok(raw); } catch (ScriptTimeoutException) @@ -164,6 +180,14 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac return VirtualTagEvalResult.Failure( $"{ScriptEvalFailureReasons.TimedOutPrefix}{_runTimeout.TotalSeconds:F1}s"); } + catch (ScriptRunRefusedException ex) + { + return VirtualTagEvalResult.Failure(ex.Message); + } + catch (ScriptQuarantinedException ex) + { + return VirtualTagEvalResult.Failure(ex.Message); + } catch (ObjectDisposedException) when (!_disposed && attempt == 0) { // S12: the apply-boundary ClearCompiledScripts disposed the evaluator between our @@ -185,7 +209,19 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac } /// - public void ClearCompiledScripts() => _cache.Clear(); + public void ClearCompiledScripts() + { + _cache.Clear(); + // The compile-cache clear IS the deploy boundary for this evaluator: a redeploy lifts every quarantine + // (an edited expression already lifts its own through the guard's source identity). + _guard.ResetAll(); + } + + /// + public void ResetQuarantines() => _guard.ResetAll(); + + /// + public IReadOnlyCollection QuarantinedScripts => _guard.QuarantinedKeys; /// Disposes the evaluator and cleans up all cached script evaluators. public void Dispose() diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs index d00dfb36..f97fa387 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs @@ -3,6 +3,7 @@ using Akka.Event; using ZB.MOM.WW.OtOpcUa.Commons.Engines; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; @@ -112,6 +113,11 @@ public sealed class VirtualTagHostActor : ReceiveActor (_evaluator as IScriptCacheOwner)?.ClearCompiledScripts(); } + // Every apply is a deploy, and a deploy lifts the run guard's quarantines unconditionally — "the + // operator redeployed to retry" is one of the two documented ways out of a quarantine (the other, + // an edited expression, the guard detects on its own). Cheap on a no-op redeploy (an empty map). + (_evaluator as IScriptQuarantineOwner)?.ResetQuarantines(); + var desired = new HashSet(msg.Plans.Select(p => p.VirtualTagId), StringComparer.Ordinal); // Stop + forget children whose vtagId is no longer desired. Stopping the child triggers its diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs new file mode 100644 index 00000000..4074d0d9 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs @@ -0,0 +1,105 @@ +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests; + +/// +/// runs every predicate through a keyed by +/// alarm id. A runaway predicate used to leak one pool thread per upstream push (each re-evaluation +/// spawned another abandoned run); now the orphan is one, later pushes are refused and the alarm is +/// quarantined — frozen at its prior state and annotated Bad quality so the operator can see it — while a +/// sibling alarm on the same engine keeps evaluating. A reload lifts the quarantine. +/// +[Trait("Category", "Unit")] +public sealed class ScriptedAlarmEngineRunGuardTests +{ + private const uint StatusBadInternalError = 0x80020000u; + + private const string SpinPredicate = + "var end = System.DateTime.UtcNow.AddSeconds(6); while (System.DateTime.UtcNow < end) { } return false;"; + + private static ScriptedAlarmDefinition Alarm(string id, string predicate) => + new(AlarmId: id, EquipmentPath: "Plant/Line1/Reactor", AlarmName: id, Kind: AlarmKind.AlarmCondition, + Severity: AlarmSeverity.High, MessageTemplate: "condition", PredicateScriptSource: predicate); + + private static ScriptedAlarmEngine Build(FakeUpstream up, ScriptingOptions? scripting = null) + { + var logger = new LoggerConfiguration().CreateLogger(); + // 500 ms, not the 60 ms the guard's own tests use: the HEALTHY alarm's first run is JIT-cold and + // must finish inside the budget on a loaded CI box, or the positive control itself times out. The + // spin predicate outlives this by 12x, so the runaway alarm still cannot finish first. + return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger, + scriptTimeout: TimeSpan.FromMilliseconds(500), scripting: scripting); + } + + [Fact] + public async Task Runaway_predicate_leaks_one_orphan_is_quarantined_annotated_Bad_and_a_sibling_keeps_evaluating() + { + var up = new FakeUpstream(); + up.Set("Temp", 50); + using var eng = Build(up); + var events = new List(); + eng.OnEvent += (_, e) => { lock (events) events.Add(e); }; + + // The load itself evaluates once (startup recovery) — that is the first, abandoned run. + await eng.LoadAsync( + [Alarm("runaway", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}"), + Alarm("healthy", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + + // Two more pushes: refused (orphan still running), then quarantined. + up.Push("Temp", 60); + await WaitForAsync(() => eng.RunGuard.RefusedWhileRunningCount >= 1); + up.Push("Temp", 70); + await WaitForAsync(() => eng.RunGuard.IsQuarantined("runaway", out _)); + + eng.RunGuard.StartedRunCount.ShouldBe(1 + 3, "one dispatch for the runaway alarm, one per evaluation for the healthy one"); + eng.RunGuard.InFlightCount.ShouldBe(1); + + // The frozen condition is annotated Bad out of band — the operator sees WHY it is not moving. + ScriptedAlarmEvent[] snapshot; + lock (events) snapshot = events.ToArray(); + var quality = snapshot.Where(e => e.AlarmId == "runaway" && e.Emission == EmissionKind.QualityChanged).ToArray(); + quality.ShouldNotBeEmpty(); + quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError); + eng.GetState("runaway")!.Active.ShouldBe(AlarmActiveState.Inactive, "state is held, never invented"); + + // Positive control: the healthy sibling still transitions on the same engine. + up.Push("Temp", 150); + await WaitForAsync(() => eng.GetState("healthy")!.Active == AlarmActiveState.Active); + } + + [Fact] + public async Task A_reload_lifts_the_quarantine() + { + var up = new FakeUpstream(); + up.Set("Temp", 50); + using var eng = Build(up, new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }); + + await eng.LoadAsync([Alarm("a", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}")], + TestContext.Current.CancellationToken); + eng.RunGuard.IsQuarantined("a", out _).ShouldBeTrue(); + + // The operator fixed the predicate and redeployed. + await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + eng.RunGuard.IsQuarantined("a", out _).ShouldBeFalse(); + + up.Push("Temp", 150); + await WaitForAsync(() => eng.GetState("a")!.Active == AlarmActiveState.Active); + } + + private static async Task WaitForAsync(Func cond, int timeoutMs = 10_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (DateTime.UtcNow < deadline) + { + if (cond()) return; + await Task.Delay(25); + } + throw new TimeoutException("Condition did not become true in time"); + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs new file mode 100644 index 00000000..974e6a7c --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs @@ -0,0 +1,306 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; + +/// +/// — the per-script run quarantine shared by the virtual-tag evaluator, the +/// Calculation evaluator and the scripted-alarm engine. A timed-out script is abandoned, not stopped, so +/// before the guard every re-trigger of a runaway script leaked one more thread-pool thread; the guard +/// refuses to start a run while the previous one is still executing, quarantines a script after three +/// consecutive non-completions, and quarantines a script whose single run allocates past the budget. +/// +[Trait("Category", "Unit")] +public sealed class ScriptRunGuardTests +{ + /// + /// A CPU-bound script that outlives every timeout in this file but still ENDS (so the orphaned pool + /// thread it leaves behind frees itself after the test). System.DateTime is outside the sandbox + /// deny-list. + /// + private const string SpinScript = + """ + var end = System.DateTime.UtcNow.AddSeconds(6); + while (System.DateTime.UtcNow < end) { } + return 1; + """; + + private const string FastScript = "return 42;"; + + private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(60); + + private static ScriptEvaluator Compile(string source) => + ScriptEvaluator.Compile(source); + + private static Task Run(ScriptRunGuard guard, string key, string source, + ScriptEvaluator evaluator, TimeSpan? timeout = null, CancellationToken ct = default) => + guard.RunAsync(key, source, evaluator, new FakeScriptContext(), timeout ?? TimeSpan.FromSeconds(5), ct); + + [Fact] + public async Task Completed_run_returns_the_value_and_releases_the_in_flight_slot() + { + var guard = new ScriptRunGuard(); + using var fast = Compile(FastScript); + + var value = await Run(guard, "vt-1", FastScript, fast, ct: TestContext.Current.CancellationToken); + + value.ShouldBe(42); + guard.InFlightCount.ShouldBe(0); + guard.StartedRunCount.ShouldBe(1); + guard.IsQuarantined("vt-1", out _).ShouldBeFalse(); + } + + [Fact] + public async Task A_timed_out_run_throws_ScriptTimeoutException_and_keeps_the_slot_held_by_the_orphan() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + + await Should.ThrowAsync(() => + Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken)); + + // The abandoned script is still executing on its pool thread — the guard knows, the caller does not. + guard.InFlightCount.ShouldBe(1); + guard.StartedRunCount.ShouldBe(1); + } + + [Fact] + public async Task A_run_while_the_previous_orphan_is_still_executing_is_refused_without_starting_another() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + await Should.ThrowAsync(() => + Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken)); + + var ex = await Should.ThrowAsync(() => + Run(guard, "vt-1", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken)); + + ex.Message.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason); + guard.StartedRunCount.ShouldBe(1, "the refusal must not dispatch a second Task.Run"); + guard.InFlightCount.ShouldBe(1); + guard.RefusedWhileRunningCount.ShouldBe(1); + } + + [Fact] + public async Task Three_consecutive_non_completions_quarantine_the_script_and_a_sibling_keeps_evaluating() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + using var fast = Compile(FastScript); + var ct = TestContext.Current.CancellationToken; + + // 1: times out, 2: refused (orphan still running), 3: refused → threshold reached → quarantined. + await Should.ThrowAsync(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct)); + await Should.ThrowAsync(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct)); + var quarantined = await Should.ThrowAsync(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct)); + + quarantined.Message.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix); + quarantined.Message.ShouldContain("3 consecutive"); + guard.IsQuarantined("bad", out var reason).ShouldBeTrue(); + reason.ShouldNotBeNull(); + guard.QuarantinedCount.ShouldBe(1); + + // A quarantined script is never dispatched again — the 4th attempt refuses before Task.Run. + await Should.ThrowAsync(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct)); + guard.StartedRunCount.ShouldBe(1); + + // Positive control: a healthy sibling on the SAME guard still runs — the quarantine is per key. + (await Run(guard, "good", FastScript, fast, ct: ct)).ShouldBe(42); + guard.IsQuarantined("good", out _).ShouldBeFalse(); + guard.StartedRunCount.ShouldBe(2); + } + + /// + /// The thread-leak regression. Pre-guard, a while(true) script on a 1 Hz trigger leaked one + /// pool thread per trigger — 30 orphans after 30 triggers, trending toward pool starvation. Now the + /// first trigger's orphan is the only one, every later trigger is refused or quarantined, and a healthy + /// sibling keeps evaluating throughout (the positive control an absence assertion needs). + /// + /// + /// The load-bearing assertions are on the guard's own bookkeeping ( + /// / ): moves with + /// every parallel test class on a loaded machine, so it is kept as a coarse secondary check only — + /// it must stay far below the 30 the leak would have produced. + /// + [Fact] + public async Task Runaway_script_leaves_at_most_one_orphan_after_30_triggers_while_a_sibling_keeps_evaluating() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + using var fast = Compile(FastScript); + var ct = TestContext.Current.CancellationToken; + var threadsBefore = ThreadPool.ThreadCount; + + var healthyResults = new List(); + for (var trigger = 0; trigger < 30; trigger++) + { + try { await Run(guard, "runaway", SpinScript, spin, ShortTimeout, ct); } + catch (ScriptTimeoutException) { } + catch (ScriptRunRefusedException) { } + catch (ScriptQuarantinedException) { } + + healthyResults.Add(await Run(guard, "healthy", FastScript, fast, ct: ct)); + } + + guard.InFlightCount.ShouldBe(1, "exactly one orphaned run — the first trigger's — may be executing"); + guard.StartedRunCount.ShouldBe(1 + 30, "one dispatch for the runaway script, thirty for the healthy sibling"); + guard.IsQuarantined("runaway", out _).ShouldBeTrue(); + healthyResults.ShouldAllBe(v => v == 42); + healthyResults.Count.ShouldBe(30); + + var threadDelta = ThreadPool.ThreadCount - threadsBefore; + threadDelta.ShouldBeLessThan(15, "thirty leaked orphans would have grown the pool by ~30 threads"); + } + + [Fact] + public async Task A_source_change_clears_the_quarantine_and_abandons_the_old_orphan_slot() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + using var fast = Compile(FastScript); + var ct = TestContext.Current.CancellationToken; + + await Should.ThrowAsync(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct)); + await Should.ThrowAsync(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct)); + await Should.ThrowAsync(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct)); + guard.IsQuarantined("vt", out _).ShouldBeTrue(); + + // The operator fixed the script: a different source identity under the same key starts clean. The + // old orphan is a sunk cost — it keeps running, but it must not block the FIXED script. + (await Run(guard, "vt", FastScript, fast, ct: ct)).ShouldBe(42); + guard.IsQuarantined("vt", out _).ShouldBeFalse(); + guard.InFlightCount.ShouldBe(1, "the abandoned orphan is still counted until it finishes"); + } + + [Fact] + public async Task Reset_and_ResetAll_clear_a_quarantine() + { + var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }); + using var spin = Compile(SpinScript); + using var fast = Compile(FastScript); + var ct = TestContext.Current.CancellationToken; + + await Should.ThrowAsync(() => Run(guard, "a", SpinScript, spin, ShortTimeout, ct)); + await Should.ThrowAsync(() => Run(guard, "b", SpinScript, spin, ShortTimeout, ct)); + guard.QuarantinedKeys.ShouldBe(new[] { "a", "b" }, ignoreOrder: true); + + guard.Reset("a"); + guard.IsQuarantined("a", out _).ShouldBeFalse(); + guard.IsQuarantined("b", out _).ShouldBeTrue(); + + guard.ResetAll(); + guard.IsQuarantined("b", out _).ShouldBeFalse(); + // The key is usable again (a fresh slot: the orphan from the quarantined run no longer blocks it). + (await Run(guard, "b", FastScript, fast, ct: ct)).ShouldBe(42); + } + + [Fact] + public async Task A_run_that_allocates_past_the_budget_is_quarantined() + { + var guard = new ScriptRunGuard(new ScriptingOptions { MaxAllocBytesPerRun = 1024 * 1024 }); + // Writes into every array so the JIT cannot fold `new byte[N].Length` to a constant and drop the + // allocation as dead code (which is exactly what it does to the naive one-liner). + const string hungry = + "long total = 0; for (var i = 0; i < 32; i++) { var a = new byte[256 * 1024]; a[i] = 1; total += a[i] + a.Length; } return (int)total;"; + using var evaluator = Compile(hungry); + using var fast = Compile(FastScript); + var ct = TestContext.Current.CancellationToken; + + var ex = await Should.ThrowAsync(() => Run(guard, "hog", hungry, evaluator, ct: ct)); + ex.Message.ShouldContain("allocat"); + guard.IsQuarantined("hog", out var reason).ShouldBeTrue(); + reason.ShouldNotBeNull(); + reason.ShouldContain("allocat"); + + // Never dispatched again, and the slot was released (the run DID complete). + await Should.ThrowAsync(() => Run(guard, "hog", hungry, evaluator, ct: ct)); + guard.StartedRunCount.ShouldBe(1); + guard.InFlightCount.ShouldBe(0); + + // Positive control: a modest allocation under the budget is fine. + (await Run(guard, "ok", FastScript, fast, ct: ct)).ShouldBe(42); + } + + [Fact] + public async Task A_disabled_allocation_budget_never_quarantines() + { + var guard = new ScriptRunGuard(new ScriptingOptions { MaxAllocBytesPerRun = 0 }); + // Writes into every array so the JIT cannot fold `new byte[N].Length` to a constant and drop the + // allocation as dead code (which is exactly what it does to the naive one-liner). + const string hungry = + "long total = 0; for (var i = 0; i < 32; i++) { var a = new byte[256 * 1024]; a[i] = 1; total += a[i] + a.Length; } return (int)total;"; + using var evaluator = Compile(hungry); + + (await Run(guard, "hog", hungry, evaluator, ct: TestContext.Current.CancellationToken)).ShouldBe(32 * (256 * 1024 + 1)); + guard.IsQuarantined("hog", out _).ShouldBeFalse(); + } + + [Fact] + public async Task A_script_that_throws_propagates_unwrapped_and_counts_as_a_completion() + { + var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 2 }); + const string throwing = "throw new System.InvalidOperationException(\"boom\");"; + using var thrower = Compile(throwing); + // A spin that outlives the timeout but finishes quickly, so the slot frees up between attempts. + const string briefSpin = + """ + var end = System.DateTime.UtcNow.AddMilliseconds(250); + while (System.DateTime.UtcNow < end) { } + return 1; + """; + using var spin = Compile(briefSpin); + var ct = TestContext.Current.CancellationToken; + + // timeout #1 + await Should.ThrowAsync(() => Run(guard, "k", briefSpin, spin, ShortTimeout, ct)); + await WaitForIdle(guard); + + // A throw is a completion — the script returned control — so the consecutive counter resets… + var ex = await Should.ThrowAsync(() => Run(guard, "k", throwing, thrower, ct: ct)); + ex.Message.ShouldBe("boom"); + guard.InFlightCount.ShouldBe(0); + + // …which is why this second timeout does NOT reach the threshold of 2. + await Should.ThrowAsync(() => Run(guard, "k", briefSpin, spin, ShortTimeout, ct)); + guard.IsQuarantined("k", out _).ShouldBeFalse(); + await WaitForIdle(guard); + } + + [Fact] + public async Task Caller_cancellation_wins_over_the_timeout() + { + var guard = new ScriptRunGuard(); + using var spin = Compile(SpinScript); + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(80)); + + await Should.ThrowAsync(() => + Run(guard, "k", SpinScript, spin, TimeSpan.FromSeconds(5), cts.Token)); + guard.IsQuarantined("k", out _).ShouldBeFalse(); + } + + [Fact] + public async Task Quarantine_is_observable_through_the_event_and_the_counter() + { + var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }); + using var spin = Compile(SpinScript); + var seen = new List<(string Key, string Reason)>(); + guard.Quarantined += (_, e) => seen.Add((e.Key, e.Reason)); + + await Should.ThrowAsync(() => + Run(guard, "k", SpinScript, spin, ShortTimeout, TestContext.Current.CancellationToken)); + + seen.ShouldHaveSingleItem(); + seen[0].Key.ShouldBe("k"); + seen[0].Reason.ShouldContain("consecutive"); + guard.QuarantinedCount.ShouldBe(1); + } + + private static async Task WaitForIdle(ScriptRunGuard guard) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (guard.InFlightCount != 0 && DateTime.UtcNow < deadline) + await Task.Delay(10); + guard.InFlightCount.ShouldBe(0, "the brief spin must have finished"); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs new file mode 100644 index 00000000..85943f16 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs @@ -0,0 +1,86 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests; + +/// +/// runs every compiled script through a +/// keyed by calc-tag RawPath. A timed-out run is abandoned on its pool thread; before the guard, every +/// later trigger of the same tag stacked another orphan behind it (one leaked thread per trigger). Now +/// the next trigger is refused ("previous evaluation still running"), the third consecutive +/// non-completion quarantines the tag, and a healthy sibling on the same evaluator keeps computing. +/// +public sealed class CalculationEvaluatorRunGuardTests +{ + private static ScriptRootLogger NoOpScriptRoot() => new(new LoggerConfiguration().CreateLogger()); + + /// Outlives the 60 ms budget by a wide margin but still ends, so the orphan frees itself. + private const string SpinScript = + "var end = System.DateTime.UtcNow.AddSeconds(6); while (System.DateTime.UtcNow < end) { } return 0;"; + + private static readonly Dictionary NoDeps = new(); + + [Fact] + public async Task A_trigger_during_the_previous_orphan_is_refused_then_the_tag_is_quarantined_and_a_sibling_keeps_computing() + { + using var sut = new CalculationEvaluator( + NullLogger.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60)); + var ct = TestContext.Current.CancellationToken; + + var first = await sut.EvaluateAsync("calc/loop", SpinScript, Snap(NoDeps), ct).WaitAsync(TimeSpan.FromSeconds(30), ct); + first.Success.ShouldBeFalse(); + first.Reason.ShouldNotBeNull(); + first.Reason.ShouldContain("timed out"); + + var second = await sut.EvaluateAsync("calc/loop", SpinScript, Snap(NoDeps), ct); + second.Success.ShouldBeFalse(); + second.Reason.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason); + + var third = await sut.EvaluateAsync("calc/loop", SpinScript, Snap(NoDeps), ct); + third.Success.ShouldBeFalse(); + third.Reason.ShouldNotBeNull(); + third.Reason.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix); + sut.QuarantinedScripts.ShouldBe(new[] { "calc/loop" }); + + // Exactly one run was ever dispatched for the runaway tag — the leak is one orphan, not one per trigger. + sut.RunGuard.StartedRunCount.ShouldBe(1); + sut.RunGuard.InFlightCount.ShouldBe(1); + + // Positive control: a sibling tag on the same evaluator still computes. + var sibling = await sut.EvaluateAsync("calc/ok", "return 42;", Snap(NoDeps), ct); + sibling.Success.ShouldBeTrue(sibling.Reason); + sibling.Value.ShouldBe(42); + } + + [Fact] + public async Task ClearCompiledScripts_lifts_the_quarantine_as_a_redeploy_does() + { + using var sut = new CalculationEvaluator( + NullLogger.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60), + new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }); + var ct = TestContext.Current.CancellationToken; + + var r = await sut.EvaluateAsync("calc/loop", SpinScript, Snap(NoDeps), ct).WaitAsync(TimeSpan.FromSeconds(30), ct); + r.Reason.ShouldNotBeNull(); + r.Reason.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix); + + sut.ClearCompiledScripts(); + + sut.QuarantinedScripts.ShouldBeEmpty(); + // The fixed script (new source) runs; the old orphan is a sunk cost that no longer blocks the tag. + var fixedRun = await sut.EvaluateAsync("calc/loop", "return 7;", Snap(NoDeps), ct); + fixedRun.Success.ShouldBeTrue(fixedRun.Reason); + fixedRun.Value.ShouldBe(7); + } + + private static Dictionary Snap(Dictionary deps) + { + var now = DateTime.UtcNow; + var d = new Dictionary(StringComparer.Ordinal); + foreach (var kv in deps) d[kv.Key] = new Core.Abstractions.DataValueSnapshot(kv.Value, 0u, now, now); + return d; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs new file mode 100644 index 00000000..f0de4999 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs @@ -0,0 +1,66 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Host.Engines; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// runs every compiled expression through a +/// keyed by virtual-tag id: a runaway expression leaks one orphaned pool +/// thread, not one per trigger, the tag is quarantined after three consecutive non-completions (surfacing +/// as a Failure the actor publishes Bad for), and the quarantine lifts at the deploy boundary. +/// +public sealed class RoslynVirtualTagEvaluatorRunGuardTests +{ + private static ScriptRootLogger NoOpScriptRoot() => new(new LoggerConfiguration().CreateLogger()); + + private const string SpinScript = + "var end = System.DateTime.UtcNow.AddSeconds(6); while (System.DateTime.UtcNow < end) { } return 0;"; + + private static readonly Dictionary NoDeps = new(); + + [Fact] + public void Runaway_expression_is_refused_then_quarantined_and_a_sibling_keeps_evaluating() + { + using var sut = new RoslynVirtualTagEvaluator( + NullLogger.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60)); + + var first = sut.Evaluate("vt-loop", SpinScript, NoDeps); + first.Success.ShouldBeFalse(); + first.Reason.ShouldNotBeNull(); + first.Reason.ShouldContain("timed out"); + + sut.Evaluate("vt-loop", SpinScript, NoDeps).Reason.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason); + + var third = sut.Evaluate("vt-loop", SpinScript, NoDeps); + third.Reason.ShouldNotBeNull(); + third.Reason.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix); + sut.QuarantinedScripts.ShouldBe(new[] { "vt-loop" }); + sut.RunGuard.StartedRunCount.ShouldBe(1, "one orphan, not one per trigger"); + + var sibling = sut.Evaluate("vt-ok", "return 42;", NoDeps); + sibling.Success.ShouldBeTrue(sibling.Reason); + sibling.Value.ShouldBe(42); + } + + [Fact] + public void ResetQuarantines_lifts_the_quarantine_as_a_redeploy_does() + { + using var sut = new RoslynVirtualTagEvaluator( + NullLogger.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60), + new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 }); + + sut.Evaluate("vt-loop", SpinScript, NoDeps).Reason!.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix); + sut.QuarantinedScripts.ShouldNotBeEmpty(); + + sut.ResetQuarantines(); + + sut.QuarantinedScripts.ShouldBeEmpty(); + var fixedRun = sut.Evaluate("vt-loop", "return 7;", NoDeps); + fixedRun.Success.ShouldBeTrue(fixedRun.Reason); + fixedRun.Value.ShouldBe(7); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorTests.cs index 004b7165..47ac617a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorTests.cs @@ -489,12 +489,18 @@ public sealed class RoslynVirtualTagEvaluatorTests // most evaluations then hit the warm cache instead of recompiling. Neither the cadence nor the // iteration count is a correctness knob any more — the assertions below hold for any number of // clears landing inside a single evaluation, and for any number of evaluations completed. - var evalLoops = Enumerable.Range(0, loopCount).Select(_ => Task.Run(() => + // One virtual-tag id PER LOOP, sharing the expression. The race under test lives in the compile + // cache, which is keyed by source, so twelve ids on one expression contend for exactly the entry + // twelve copies of one id did — while twelve concurrent evaluations of ONE id is a shape production + // never produces (one VirtualTagActor per id serialises them) and exactly the shape the per-tag + // ScriptRunGuard refuses ("previous evaluation still running", then quarantine). + var evalLoops = Enumerable.Range(0, loopCount).Select(loop => Task.Run(() => { + var vtagId = $"vt-race-{loop}"; var done = 0; for (var i = 0; i < MaxIterationsPerLoop && Environment.TickCount64 < stopAtMs; i++) { - var r = sut.Evaluate("vt-race", expr, new Dictionary { ["a"] = i }); + var r = sut.Evaluate(vtagId, expr, new Dictionary { ["a"] = i }); done++; if (!r.Success) {