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 9438132a..ae6352f5 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs @@ -76,7 +76,7 @@ public sealed class ScriptedAlarmEngine : IDisposable /// /// 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 (a redeploy). /// 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) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs index 8f1acefd..dc308ce6 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptRunGuard.cs @@ -24,8 +24,10 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; /// (), 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 +/// consecutive timeouts (only a timeout is a strike; a refusal is a Bad publish plus its counter +/// and leaves the streak untouched — on a 1 Hz trigger with a 2 s budget one 4 s overrun is a +/// timeout plus two refusals and must not quarantine; a run that returns — a value or a throw — +/// resets the streak): 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 @@ -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("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 } } - /// Record a non-completion; returns true when it tipped the key into quarantine. - private bool NoteNonCompletion(string key, KeyState state, string what) + /// Record a timeout strike; returns true when it tipped the key into quarantine. A completed run + /// (value or throw) resets the streak; a refusal neither counts nor resets. + private bool NoteTimeout(string key, KeyState state, TimeSpan timeout) { - var count = Interlocked.Increment(ref state.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 /// /// Thrown by 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. /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs index ba1399fa..f101a0e1 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptingOptions.cs @@ -33,10 +33,11 @@ public sealed class ScriptingOptions 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 + /// 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. /// public int QuarantineAfterConsecutiveTimeouts { get; set; } = DefaultQuarantineAfterConsecutiveTimeouts; } 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 349a0cf1..1a589d65 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriverFactoryExtensions.cs @@ -107,7 +107,7 @@ public static class CalculationDriverFactoryExtensions /// 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). + /// Optional run-guard quarantine threshold: consecutive timeouts (refusals do not count) 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/CalculationEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs index ba4931c7..e47cc745 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationEvaluator.cs @@ -42,7 +42,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation; /// 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 +/// timeouts (a refusal is Bad but not a strike) 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 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 c6823c41..f0bbd150 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Engines/RoslynVirtualTagEvaluator.cs @@ -35,7 +35,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines; /// 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 +/// timeouts (a refusal is Bad but not a strike) 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. 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 index 4074d0d9..ab2fde04 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineRunGuardTests.cs @@ -9,8 +9,8 @@ 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 +/// 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. /// [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); } + /// 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. + [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(); + 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() { 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 index 974e6a7c..f60045b9 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptRunGuardTests.cs @@ -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. /// [Trait("Category", "Unit")] public sealed class ScriptRunGuardTests @@ -28,6 +28,14 @@ public sealed class ScriptRunGuardTests private const string FastScript = "return 42;"; + /// Outlives the 60 ms budget but finishes within ~250 ms, so the slot frees up between attempts. + 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 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(() => 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)); + // 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(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct)); + await WaitForIdle(guard); + await Should.ThrowAsync(() => Run(guard, "bad", BriefSpinScript, spin, ShortTimeout, ct)); + await WaitForIdle(guard); + var quarantined = await Should.ThrowAsync(() => 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(() => Run(guard, "bad", SpinScript, spin, ShortTimeout, ct)); - guard.StartedRunCount.ShouldBe(1); + await Should.ThrowAsync(() => 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); + } + + /// + /// 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. + /// + [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(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct)); + for (var i = 0; i < 5; i++) + await Should.ThrowAsync(() => 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(() => Run(guard, "k", BriefSpinScript, spin, ShortTimeout, ct)); + guard.IsQuarantined("k", out _).ShouldBeTrue(); + await WaitForIdle(guard); } /// /// 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 + /// 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). /// /// @@ -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(() => 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(); 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 index 85943f16..679e7f71 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationEvaluatorRunGuardTests.cs @@ -10,8 +10,8 @@ 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. +/// 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. /// public sealed class CalculationEvaluatorRunGuardTests { @@ -24,7 +24,7 @@ public sealed class CalculationEvaluatorRunGuardTests 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() + public async Task Triggers_during_the_previous_orphan_are_refused_not_quarantined_and_a_sibling_keeps_computing() { using var sut = new CalculationEvaluator( NullLogger.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); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs index f0de4999..cd18ae5e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/RoslynVirtualTagEvaluatorRunGuardTests.cs @@ -10,7 +10,7 @@ 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 +/// thread, not one per trigger, later triggers are refused, the tag is quarantined after three consecutive timeouts (surfacing /// as a Failure the actor publishes Bad for), and the quarantine lifts at the deploy boundary. /// public sealed class RoslynVirtualTagEvaluatorRunGuardTests @@ -23,7 +23,7 @@ public sealed class RoslynVirtualTagEvaluatorRunGuardTests private static readonly Dictionary 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.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);