fix(scripting): only timeouts are quarantine strikes; alarm Bad annotation once per bucket change (fix round 1 I1, I3)
- I1 ScriptRunGuard: a run refused because the previous one is still executing is a Bad publish plus otopcua.script.run_refused, and no longer a quarantine strike. Only a timeout (and an allocation breach) counts; a refusal neither adds to nor resets the timeout streak; a completed run (value or throw) resets it. On a 1 Hz trigger with a 2 s budget a single 4 s overrun is one timeout plus two refusals and must not quarantine. Tests updated across the guard, the alarm engine, the Calculation evaluator and the VT evaluator, plus Refusals_neither_count_toward_nor_reset_the_timeout_streak. - I3 ScriptedAlarmEngine: the quality bucket is tracked only after the predicate completes (or in the inputs-not-ready path), so the tracked bucket is the last one annotated. A refused or quarantined predicate used to re-track the input bucket Good before the attempt and then flip it Bad, emitting QualityChanged on every push. The run-guard tests now count emissions (exactly 1 across N pushes, refused and quarantined cases), with a healthy sibling's dispatch count as the positive control that every push was processed. Both were red before the fix. Refs #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -76,7 +76,7 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
/// <summary>
|
||||
/// Per-alarm run quarantine (keyed by alarm id, source identity = predicate source). Refuses a
|
||||
/// predicate run while the alarm's previous (timed-out, abandoned) run is still executing and
|
||||
/// quarantines after three consecutive non-completions or an allocation-budget breach — one orphan
|
||||
/// quarantines after three consecutive timeouts (a refusal is not a strike) or an allocation-budget breach — one orphan
|
||||
/// per alarm instead of one per upstream push. Reset on every <see cref="LoadAsync"/> (a redeploy).
|
||||
/// </summary>
|
||||
private readonly ScriptRunGuard _guard;
|
||||
@@ -798,7 +798,6 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
// short-circuit so an outright-Bad input is still observed. A bucket change with no state
|
||||
// transition is delivered out of band as a QualityChanged emission (see below).
|
||||
var worstStatus = WorstInputStatus(scratch.ReadCache);
|
||||
var qualityBucketChanged = TrackQualityBucket(state.Definition.AlarmId, worstStatus);
|
||||
|
||||
// Cold-start guard — skip the predicate when any referenced upstream tag has no
|
||||
// cached value yet (the upstream subscription hasn't delivered its first push).
|
||||
@@ -811,7 +810,7 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
// The condition is frozen (can't trust its state), but its source quality just changed
|
||||
// bucket — annotate it out of band so a comms-lost / Bad-input scripted condition reports
|
||||
// Bad, mirroring the native OT path.
|
||||
if (qualityBucketChanged)
|
||||
if (TrackQualityBucket(state.Definition.AlarmId, worstStatus))
|
||||
pendingEmissions.Add(BuildQualityEmission(state, seed, worstStatus));
|
||||
return seed;
|
||||
}
|
||||
@@ -851,6 +850,12 @@ public sealed class ScriptedAlarmEngine : IDisposable
|
||||
}
|
||||
scratch.LastNonCompletion = null;
|
||||
|
||||
// Track the bucket only once the predicate has COMPLETED (I3). The tracked bucket is the last
|
||||
// bucket this alarm was annotated with; tracking the input bucket before the attempt flipped it to
|
||||
// Good on every push, so a refused / quarantined predicate (which annotates Bad) re-emitted
|
||||
// QualityChanged on every push instead of once per real bucket change.
|
||||
var qualityBucketChanged = TrackQualityBucket(state.Definition.AlarmId, worstStatus);
|
||||
|
||||
var result = Part9StateMachine.ApplyPredicate(seed, predicateTrue, nowUtc);
|
||||
var transition = result.Emission != EmissionKind.None
|
||||
? BuildEmission(state, result.State, result.Emission, worstStatus)
|
||||
|
||||
@@ -24,8 +24,10 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
/// <see cref="ScriptRunRefusedException"/> (<see cref="PreviousRunStillRunningReason"/>), counted on
|
||||
/// <see cref="RefusedWhileRunningCount"/>. One orphan per script, never one per trigger.</item>
|
||||
/// <item><b>Quarantines after <see cref="ScriptingOptions.QuarantineAfterConsecutiveTimeouts"/>
|
||||
/// consecutive non-completions</b> (timeouts and still-running refusals both count; a run that
|
||||
/// returns — a value or a throw — resets the count): no further dispatch, every attempt throws
|
||||
/// consecutive timeouts</b> (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): no further dispatch, every attempt throws
|
||||
/// <see cref="ScriptQuarantinedException"/>, one Error log, one <c>otopcua.script.quarantined</c>
|
||||
/// increment, one <see cref="Quarantined"/> event. Lifted by a source change (the operator fixed the
|
||||
/// script — a new source identity under the same key starts clean and the old orphan no longer blocks
|
||||
@@ -177,10 +179,12 @@ public sealed class ScriptRunGuard
|
||||
// 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<string, object?>("script", key));
|
||||
if (NoteNonCompletion(key, state, "refused: previous evaluation still running"))
|
||||
throw new ScriptQuarantinedException(key, state.QuarantineReason!);
|
||||
throw new ScriptRunRefusedException(key);
|
||||
}
|
||||
|
||||
@@ -241,7 +245,7 @@ public sealed class ScriptRunGuard
|
||||
// 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"))
|
||||
if (NoteTimeout(key, state, timeout))
|
||||
throw new ScriptQuarantinedException(key, state.QuarantineReason!);
|
||||
throw new ScriptTimeoutException(timeout);
|
||||
}
|
||||
@@ -252,11 +256,11 @@ public sealed class ScriptRunGuard
|
||||
catch (Exception)
|
||||
{
|
||||
// The script returned control (by throwing) — that is a completion for quarantine purposes.
|
||||
Interlocked.Exchange(ref state.ConsecutiveNonCompletions, 0);
|
||||
Interlocked.Exchange(ref state.ConsecutiveTimeouts, 0);
|
||||
throw;
|
||||
}
|
||||
|
||||
Interlocked.Exchange(ref state.ConsecutiveNonCompletions, 0);
|
||||
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
|
||||
@@ -291,13 +295,14 @@ public sealed class ScriptRunGuard
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Record a non-completion; returns true when it tipped the key into quarantine.</summary>
|
||||
private bool NoteNonCompletion(string key, KeyState state, string what)
|
||||
/// <summary>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.</summary>
|
||||
private bool NoteTimeout(string key, KeyState state, TimeSpan timeout)
|
||||
{
|
||||
var count = Interlocked.Increment(ref state.ConsecutiveNonCompletions);
|
||||
var count = Interlocked.Increment(ref state.ConsecutiveTimeouts);
|
||||
if (count < _quarantineThreshold) return false;
|
||||
Quarantine(key, state,
|
||||
$"{count} consecutive evaluation(s) did not complete (last: {what}); " +
|
||||
$"{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;
|
||||
@@ -333,7 +338,7 @@ public sealed class ScriptRunGuard
|
||||
{
|
||||
public string? SourceIdentity = sourceIdentity;
|
||||
public int InFlight;
|
||||
public int ConsecutiveNonCompletions;
|
||||
public int ConsecutiveTimeouts;
|
||||
public string? QuarantineReason;
|
||||
}
|
||||
|
||||
@@ -368,7 +373,7 @@ public sealed class ScriptRunRefusedException : Exception
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by <see cref="ScriptRunGuard.RunAsync{TContext, TResult}"/> for a quarantined script: repeated
|
||||
/// non-completion or an allocation-budget breach. Nothing is dispatched while quarantined; the seam
|
||||
/// timeouts or an allocation-budget breach. Nothing is dispatched while quarantined; the seam
|
||||
/// publishes Bad carrying the reason, which an operator clears by fixing the script (source change) or
|
||||
/// redeploying.
|
||||
/// </summary>
|
||||
|
||||
@@ -33,10 +33,11 @@ public sealed class ScriptingOptions
|
||||
public long MaxAllocBytesPerRun { get; set; } = DefaultMaxAllocBytesPerRun;
|
||||
|
||||
/// <summary>
|
||||
/// Consecutive non-completions (a run that times out, or a run refused because the previous one is
|
||||
/// still executing) after which the script is quarantined: no further runs are dispatched, every
|
||||
/// Consecutive timeouts 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.
|
||||
/// reset. Only a timeout is a strike — a run refused because the previous one is still executing is
|
||||
/// reported Bad and counted, but neither adds to nor resets the streak; a completed run (value or
|
||||
/// throw) resets it. Default 3; values below 1 are treated as 1.
|
||||
/// </summary>
|
||||
public int QuarantineAfterConsecutiveTimeouts { get; set; } = DefaultQuarantineAfterConsecutiveTimeouts;
|
||||
}
|
||||
|
||||
+1
-1
@@ -107,7 +107,7 @@ public static class CalculationDriverFactoryExtensions
|
||||
/// <summary>Optional run-guard allocation budget per script run, in bytes (default 64 MB; ≤ 0 disables).</summary>
|
||||
public long? MaxAllocBytesPerRun { get; init; }
|
||||
|
||||
/// <summary>Optional run-guard quarantine threshold: consecutive non-completions before a tag's script stops being evaluated (default 3).</summary>
|
||||
/// <summary>Optional run-guard quarantine threshold: consecutive timeouts (refusals do not count) before a tag's script stops being evaluated (default 3).</summary>
|
||||
public int? QuarantineAfterConsecutiveTimeouts { get; init; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,7 +42,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
|
||||
/// <b>Run quarantine.</b> Every compiled run goes through one <see cref="ScriptRunGuard"/> 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 <see cref="VirtualTagEvalResult.Failure"/>
|
||||
/// timeouts (a refusal is Bad but not a strike) or an allocation-budget breach — both surface as a <see cref="VirtualTagEvalResult.Failure"/>
|
||||
/// the driver maps to Bad. The driver's evaluation pump awaits <see cref="EvaluateAsync"/> rather than
|
||||
/// blocking a pool thread on the sync <see cref="Evaluate(string, string, IReadOnlyDictionary{string, DataValueSnapshot})"/>,
|
||||
/// and the guard starts the budget clock only once the pool has picked the script up, so pool starvation
|
||||
|
||||
@@ -35,7 +35,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines;
|
||||
/// <b>Run quarantine.</b> Every compiled run goes through one <see cref="ScriptRunGuard"/> 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 <see cref="VirtualTagEvalResult.Failure"/>
|
||||
/// timeouts (a refusal is Bad but not a strike) or an allocation-budget breach. Both surface as <see cref="VirtualTagEvalResult.Failure"/>
|
||||
/// so <c>VirtualTagActor</c> publishes Bad exactly as it does for a timeout. A quarantine lifts when the
|
||||
/// expression changes, on <see cref="ClearCompiledScripts"/> / <see cref="ResetQuarantines"/> (the deploy
|
||||
/// boundary), never by itself.
|
||||
|
||||
+56
-11
@@ -9,8 +9,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;
|
||||
/// <summary>
|
||||
/// <see cref="ScriptedAlarmEngine"/> runs every predicate through a <see cref="ScriptRunGuard"/> 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
|
||||
/// spawned another abandoned run); now the orphan is one, later pushes are refused (and a
|
||||
/// predicate that times out repeatedly 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.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
@@ -36,7 +36,7 @@ public sealed class ScriptedAlarmEngineRunGuardTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Runaway_predicate_leaks_one_orphan_is_quarantined_annotated_Bad_and_a_sibling_keeps_evaluating()
|
||||
public async Task Runaway_predicate_leaks_one_orphan_is_refused_annotated_Bad_once_and_a_sibling_keeps_evaluating()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("Temp", 50);
|
||||
@@ -50,20 +50,27 @@ public sealed class ScriptedAlarmEngineRunGuardTests
|
||||
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 _));
|
||||
// N more pushes while the orphan runs: every one is refused (a refusal is not a strike — I1), and
|
||||
// the Bad annotation is emitted ONCE for the whole streak, not once per push (I3).
|
||||
const int pushes = 4;
|
||||
for (var i = 0; i < pushes; i++)
|
||||
{
|
||||
up.Push("Temp", 60 + i);
|
||||
var expected = i + 1;
|
||||
await WaitForAsync(() => eng.RunGuard.RefusedWhileRunningCount >= expected);
|
||||
}
|
||||
await WaitForAsync(() => eng.RunGuard.StartedRunCount >= 1 + 1 + pushes);
|
||||
|
||||
eng.RunGuard.StartedRunCount.ShouldBe(1 + 3, "one dispatch for the runaway alarm, one per evaluation for the healthy one");
|
||||
eng.RunGuard.IsQuarantined("runaway", out _).ShouldBeFalse("refusals are not strikes");
|
||||
eng.RunGuard.StartedRunCount.ShouldBe(1 + 1 + pushes, "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.
|
||||
// The frozen condition is annotated Bad out of band — the operator sees WHY it is not moving —
|
||||
// exactly once across the whole run of non-completions.
|
||||
ScriptedAlarmEvent[] snapshot;
|
||||
lock (events) snapshot = events.ToArray();
|
||||
var quality = snapshot.Where(e => e.AlarmId == "runaway" && e.Emission == EmissionKind.QualityChanged).ToArray();
|
||||
quality.ShouldNotBeEmpty();
|
||||
quality.Length.ShouldBe(1);
|
||||
quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError);
|
||||
eng.GetState("runaway")!.Active.ShouldBe(AlarmActiveState.Inactive, "state is held, never invented");
|
||||
|
||||
@@ -72,6 +79,44 @@ public sealed class ScriptedAlarmEngineRunGuardTests
|
||||
await WaitForAsync(() => eng.GetState("healthy")!.Active == AlarmActiveState.Active);
|
||||
}
|
||||
|
||||
/// <summary>I3 — a QUARANTINED predicate is re-attempted on every push and refused before dispatch; the
|
||||
/// Bad annotation must still be emitted once, not once per push.</summary>
|
||||
[Fact]
|
||||
public async Task A_quarantined_predicate_emits_QualityChanged_once_across_many_pushes()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("Temp", 50);
|
||||
using var eng = Build(up, new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 1 });
|
||||
var events = new List<ScriptedAlarmEvent>();
|
||||
eng.OnEvent += (_, e) => { lock (events) events.Add(e); };
|
||||
|
||||
await eng.LoadAsync(
|
||||
[Alarm("q", $"var t = ctx.GetTag(\"Temp\"); {SpinPredicate}"),
|
||||
Alarm("healthy", """return (int)ctx.GetTag("Temp").Value > 1000;""")],
|
||||
TestContext.Current.CancellationToken);
|
||||
eng.RunGuard.IsQuarantined("q", out _).ShouldBeTrue();
|
||||
|
||||
// Each push re-evaluates BOTH alarms (same input). The healthy sibling's dispatch count is the
|
||||
// positive control proving every push was actually processed — so the quarantined alarm was
|
||||
// re-attempted N times, not coalesced into one.
|
||||
const int pushes = 5;
|
||||
for (var i = 0; i < pushes; i++)
|
||||
{
|
||||
var before = eng.RunGuard.StartedRunCount;
|
||||
up.Push("Temp", 60 + i);
|
||||
await WaitForAsync(() => eng.RunGuard.StartedRunCount > before);
|
||||
}
|
||||
eng.RunGuard.StartedRunCount.ShouldBe(1 + 1 + pushes, "one runaway dispatch, then healthy-only dispatches");
|
||||
await Task.Delay(300); // absence window for trailing emissions
|
||||
|
||||
ScriptedAlarmEvent[] snapshot;
|
||||
lock (events) snapshot = events.ToArray();
|
||||
var quality = snapshot.Where(e => e.AlarmId == "q" && e.Emission == EmissionKind.QualityChanged).ToArray();
|
||||
quality.Length.ShouldBe(1, "one Bad annotation for the quarantine, not one per push");
|
||||
quality[0].WorstInputStatusCode.ShouldBe(StatusBadInternalError);
|
||||
snapshot.ShouldNotContain(e => e.AlarmId == "healthy" && e.Emission == EmissionKind.QualityChanged);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_reload_lifts_the_quarantine()
|
||||
{
|
||||
|
||||
@@ -9,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;
|
||||
/// 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.
|
||||
/// consecutive timeouts (a refusal is not a strike), and quarantines a script whose single run allocates past the budget.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class ScriptRunGuardTests
|
||||
@@ -28,6 +28,14 @@ public sealed class ScriptRunGuardTests
|
||||
|
||||
private const string FastScript = "return 42;";
|
||||
|
||||
/// <summary>Outlives the 60 ms budget but finishes within ~250 ms, so the slot frees up between attempts.</summary>
|
||||
private const string BriefSpinScript =
|
||||
"""
|
||||
var end = System.DateTime.UtcNow.AddMilliseconds(250);
|
||||
while (System.DateTime.UtcNow < end) { }
|
||||
return 1;
|
||||
""";
|
||||
|
||||
private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(60);
|
||||
|
||||
private static ScriptEvaluator<FakeScriptContext, int> Compile(string source) =>
|
||||
@@ -83,17 +91,21 @@ public sealed class ScriptRunGuardTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Three_consecutive_non_completions_quarantine_the_script_and_a_sibling_keeps_evaluating()
|
||||
public async Task Three_consecutive_timeouts_quarantine_the_script_and_a_sibling_keeps_evaluating()
|
||||
{
|
||||
var guard = new ScriptRunGuard();
|
||||
using var spin = Compile(SpinScript);
|
||||
using var spin = Compile(BriefSpinScript);
|
||||
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<ScriptTimeoutException>(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct));
|
||||
await Should.ThrowAsync<ScriptRunRefusedException>(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct));
|
||||
var quarantined = await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct));
|
||||
// Three timeouts in a row — each orphan is allowed to finish so the next attempt is a real run,
|
||||
// not a refusal (a refusal is not a strike; see the next test).
|
||||
await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
await WaitForIdle(guard);
|
||||
await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
await WaitForIdle(guard);
|
||||
var quarantined = await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
await WaitForIdle(guard);
|
||||
|
||||
quarantined.Message.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
|
||||
quarantined.Message.ShouldContain("3 consecutive");
|
||||
@@ -102,19 +114,46 @@ public sealed class ScriptRunGuardTests
|
||||
guard.QuarantinedCount.ShouldBe(1);
|
||||
|
||||
// A quarantined script is never dispatched again — the 4th attempt refuses before Task.Run.
|
||||
await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct));
|
||||
guard.StartedRunCount.ShouldBe(1);
|
||||
await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
guard.StartedRunCount.ShouldBe(3);
|
||||
|
||||
// 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);
|
||||
guard.StartedRunCount.ShouldBe(4);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A refusal is NOT a strike (review fix round 1, I1): 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. A refusal is a Bad publish plus its counter, and it leaves the timeout
|
||||
/// streak untouched — neither adding to it nor resetting it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Refusals_neither_count_toward_nor_reset_the_timeout_streak()
|
||||
{
|
||||
var guard = new ScriptRunGuard(new ScriptingOptions { QuarantineAfterConsecutiveTimeouts = 2 });
|
||||
using var spin = Compile(BriefSpinScript);
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
|
||||
// timeout #1, then refusals while the orphan is still running — however many, no quarantine.
|
||||
await Should.ThrowAsync<ScriptTimeoutException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
for (var i = 0; i < 5; i++)
|
||||
await Should.ThrowAsync<ScriptRunRefusedException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
guard.IsQuarantined("k", out _).ShouldBeFalse("refusals are not strikes");
|
||||
guard.RefusedWhileRunningCount.ShouldBe(5);
|
||||
await WaitForIdle(guard);
|
||||
|
||||
// …and they did not RESET the streak either: the second real timeout reaches the threshold of 2.
|
||||
await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct));
|
||||
guard.IsQuarantined("k", out _).ShouldBeTrue();
|
||||
await WaitForIdle(guard);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The thread-leak regression. Pre-guard, a <c>while(true)</c> 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
|
||||
/// first trigger's orphan is the only one, every later trigger is refused, and a healthy
|
||||
/// sibling keeps evaluating throughout (the positive control an absence assertion needs).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
@@ -145,7 +184,9 @@ public sealed class ScriptRunGuardTests
|
||||
|
||||
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();
|
||||
// One timeout + 29 refusals: not quarantined (a refusal is not a strike), but never re-dispatched.
|
||||
guard.IsQuarantined("runaway", out _).ShouldBeFalse();
|
||||
guard.RefusedWhileRunningCount.ShouldBe(29);
|
||||
healthyResults.ShouldAllBe(v => v == 42);
|
||||
healthyResults.Count.ShouldBe(30);
|
||||
|
||||
@@ -156,13 +197,11 @@ public sealed class ScriptRunGuardTests
|
||||
[Fact]
|
||||
public async Task A_source_change_clears_the_quarantine_and_abandons_the_old_orphan_slot()
|
||||
{
|
||||
var guard = new ScriptRunGuard();
|
||||
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<ScriptTimeoutException>(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct));
|
||||
await Should.ThrowAsync<ScriptRunRefusedException>(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct));
|
||||
await Should.ThrowAsync<ScriptQuarantinedException>(() => Run(guard, "vt", SpinScript, spin, ShortTimeout, ct));
|
||||
guard.IsQuarantined("vt", out _).ShouldBeTrue();
|
||||
|
||||
|
||||
+13
-12
@@ -10,8 +10,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;
|
||||
/// <see cref="CalculationEvaluator"/> runs every compiled script through a <see cref="ScriptRunGuard"/>
|
||||
/// 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.
|
||||
/// every later trigger is refused ("previous evaluation still running" — Bad, but not a quarantine
|
||||
/// strike; only consecutive timeouts quarantine), and a healthy sibling on the same evaluator keeps computing.
|
||||
/// </summary>
|
||||
public sealed class CalculationEvaluatorRunGuardTests
|
||||
{
|
||||
@@ -24,7 +24,7 @@ public sealed class CalculationEvaluatorRunGuardTests
|
||||
private static readonly Dictionary<string, object?> NoDeps = new();
|
||||
|
||||
[Fact]
|
||||
public async Task A_trigger_during_the_previous_orphan_is_refused_then_the_tag_is_quarantined_and_a_sibling_keeps_computing()
|
||||
public async Task Triggers_during_the_previous_orphan_are_refused_not_quarantined_and_a_sibling_keeps_computing()
|
||||
{
|
||||
using var sut = new CalculationEvaluator(
|
||||
NullLogger<CalculationEvaluator>.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60));
|
||||
@@ -35,15 +35,16 @@ public sealed class CalculationEvaluatorRunGuardTests
|
||||
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" });
|
||||
// Every later trigger while the orphan runs is a Bad "previous evaluation still running" — a
|
||||
// refusal is not a strike, so the tag is NOT quarantined by them (fix round 1, I1).
|
||||
for (var i = 0; i < 3; i++)
|
||||
{
|
||||
var refused = await sut.EvaluateAsync("calc/loop", SpinScript, Snap(NoDeps), ct);
|
||||
refused.Success.ShouldBeFalse();
|
||||
refused.Reason.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason);
|
||||
}
|
||||
sut.QuarantinedScripts.ShouldBeEmpty();
|
||||
sut.RunGuard.RefusedWhileRunningCount.ShouldBe(3);
|
||||
|
||||
// Exactly one run was ever dispatched for the runaway tag — the leak is one orphan, not one per trigger.
|
||||
sut.RunGuard.StartedRunCount.ShouldBe(1);
|
||||
|
||||
+5
-7
@@ -10,7 +10,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;
|
||||
/// <summary>
|
||||
/// <see cref="RoslynVirtualTagEvaluator"/> runs every compiled expression through a
|
||||
/// <see cref="ScriptRunGuard"/> 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
|
||||
/// thread, not one per trigger, later triggers are refused, the tag is quarantined after three consecutive timeouts (surfacing
|
||||
/// as a <c>Failure</c> the actor publishes Bad for), and the quarantine lifts at the deploy boundary.
|
||||
/// </summary>
|
||||
public sealed class RoslynVirtualTagEvaluatorRunGuardTests
|
||||
@@ -23,7 +23,7 @@ public sealed class RoslynVirtualTagEvaluatorRunGuardTests
|
||||
private static readonly Dictionary<string, object?> NoDeps = new();
|
||||
|
||||
[Fact]
|
||||
public void Runaway_expression_is_refused_then_quarantined_and_a_sibling_keeps_evaluating()
|
||||
public void Runaway_expression_is_refused_while_its_orphan_runs_and_a_sibling_keeps_evaluating()
|
||||
{
|
||||
using var sut = new RoslynVirtualTagEvaluator(
|
||||
NullLogger<RoslynVirtualTagEvaluator>.Instance, NoOpScriptRoot(), TimeSpan.FromMilliseconds(60));
|
||||
@@ -33,12 +33,10 @@ public sealed class RoslynVirtualTagEvaluatorRunGuardTests
|
||||
first.Reason.ShouldNotBeNull();
|
||||
first.Reason.ShouldContain("timed out");
|
||||
|
||||
// Refusals, not strikes (fix round 1, I1): the tag stays Bad but is not quarantined by them.
|
||||
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.Evaluate("vt-loop", SpinScript, NoDeps).Reason.ShouldBe(ScriptRunGuard.PreviousRunStillRunningReason);
|
||||
sut.QuarantinedScripts.ShouldBeEmpty();
|
||||
sut.RunGuard.StartedRunCount.ShouldBe(1, "one orphan, not one per trigger");
|
||||
|
||||
var sibling = sut.Evaluate("vt-ok", "return 42;", NoDeps);
|
||||
|
||||
Reference in New Issue
Block a user