dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

test(script-worker): prove Shadow non-blocking causally, not by stopwatch (#798)

Browse Source 
ShadowModeTests' runaway case asserted "slowest < 250 ms" around evaluations
whose timing included cold Roslyn compile/JIT of new script text, so it failed
on a healthy published path (540-880 ms measured, both SDK baselines).

The test now puts a test-owned barrier runner (GatedShadow) between the real
ShadowScriptRunner and the real worker pool. The runaway's first shadow run is
held before it reaches the pool, then (after it runs away in the worker, the
pool answers T and the worker is KilledDeadline with the runaway as culprit)
its outcome is held before comparison. Published evaluations made while a
barrier is closed must return the in-process values; a Mirror that waited for
its shadow hangs into a 30 s liveness failure (mutation-checked). Output
correctness, pool kill/quarantine, published-side never quarantined, the single
runaway mismatch (sibling is a victim, not compared) and key_busy drops are kept.

The wall-clock number moves to an opt-in Category=Perf case
(Shadow_published_path_latency_while_the_shadow_runs_away, OTOPCUA_SCRIPT_PERF=1)
on warmed evaluators, reporting hardware and load average.

Sibling: ContainmentTests' hard-deadline case timed the runaway's first
(cold-compile) run; it now compiles the runaway's source first with an input
that does not spin.
This commit is contained in:
Joseph Doherty
2026-11-10 21:15:12 -05:00
parent b36e688f0a
commit 304d1f1e6c
3 changed files with 214 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ContainmentTests.cs
+7 -3
View File
@@ -61,10 +61,14 @@ public sealed class ContainmentTests
await using var runner = Scripts.Runner(Scripts.Options(hardGrace: grace, quarantineAfterKills: 2));
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
// Compile the runaway's source first (it only spins when told to), so the stopwatch below times the soft answer
// and the kill, not the worker's first compile of new script text (lmxopcua#798's flaw, in its runner form).
const string runaway = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (true) { } } return 0;";
(await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(0);
var worker = runner.Shards[0].Current.ShouldNotBeNull();
var sw = Stopwatch.StartNew();
var outcome = await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct);
var outcome = await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct);
var softElapsed = sw.Elapsed;
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
outcome.Reason!.ShouldStartWith("script timed out after ");
@@ -90,12 +94,12 @@ public sealed class ContainmentTests
// A second kill quarantines (QuarantineAfterKills = 2); the third trigger is refused without running.
var second = runner.Shards[0].Current!;
(await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct)).Status.ShouldBe(ScriptRunStatus.Failed);
(await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct)).Status.ShouldBe(ScriptRunStatus.Failed);
(await second.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct)).Reason.ShouldBe(WorkerExitReason.KilledDeadline);
// The strike is recorded by the shard's exit handler, which runs asynchronously to the Exited completion.
await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1,
TimeSpan.FromSeconds(5), "the second kill strike");
var refused = await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct);
var refused = await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct);
refused.Reason!.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
refused.Reason.ShouldContain("hard deadline");
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["runaway"]);
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs
+91
View File
@@ -571,4 +571,95 @@ public sealed class ScriptRunnerPerfTests
.AppendLine($"| kill latency after T+G (5 runs, ms) | {string.Join(" / ", killLatencies.Select(l => l.ToString("N0")))} | ≤ 100 ms |");
Report(report);
}
/// <summary>
/// lmxopcua#798: the wall-clock half of the Shadow non-blocking contract, moved out of the default
/// <see cref="ShadowModeTests"/> run (which now proves it causally with barriers). Published-path latency of a
/// WARMED Calculation evaluator under <c>Scripting:Runner=Shadow</c> while its shadow keeps running away in the
/// worker and being killed, against the same evaluator with no shadow. Compile/JIT is excluded by a warm-up
/// pass. The 250 ms per-evaluation ceiling the default test used to assert is asserted here, opt-in only; the
/// report records hardware and (where the OS exposes it) load average, so a MISS on a loaded box reads as such.
/// </summary>
[Fact]
public async Task Shadow_published_path_latency_while_the_shadow_runs_away()
{
RequireOptIn();
const string inWorker = "(string.Compare(\"a\", \"B\") > 0)"; // see ShadowModeTests.InWorker
const string runaway = $"if ({inWorker}) {{ while (true) {{ }} }} return (int)ctx.GetTag(\"x\").Value * 2;";
const string sibling = "return (int)ctx.GetTag(\"x\").Value + 100;";
const int count = 2_000;
var options = Scripts.Options(hardGrace: TimeSpan.FromMilliseconds(400), quarantineAfterKills: 1_000_000);
options.Runner = ScriptRunnerMode.Shadow;
await using var pool = Scripts.Runner(options, _runnerLog.CreateLogger());
using var shadow = new ShadowScriptRunner(pool, options, _runnerLog.CreateLogger());
Driver.Calculation.CalculationEvaluator Calc(IScriptRunner? runner)
=> new(Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance,
new ScriptRootLogger(new Serilog.LoggerConfiguration().CreateLogger()), TimeSpan.FromMilliseconds(300), scriptRunner: runner);
async Task<List<double>> Measure(Driver.Calculation.CalculationEvaluator evaluator)
{
for (var w = 0; w < 50; w++) // warm-up: compile + JIT both scripts in process
{
(await evaluator.EvaluateAsync("runaway", runaway, Scripts.Input("x", w), Ct)).Value.ShouldBe(w * 2);
(await evaluator.EvaluateAsync("sibling", sibling, Scripts.Input("x", w), Ct)).Value.ShouldBe(w + 100);
}
// A 5 ms trigger cadence for the phase, so shadow runs complete, run away and are killed DURING the
// measurement (back-to-back evaluations would all land behind the first shadow run, key_busy).
var samples = new List<double>(count);
var phaseEnd = Stopwatch.GetTimestamp() + (long)(Phase.TotalSeconds * Stopwatch.Frequency);
for (var n = 0; Stopwatch.GetTimestamp() < phaseEnd; n++)
{
await Task.Delay(5, Ct);
var key = n % 2 == 0 ? "runaway" : "sibling";
var t0 = Stopwatch.GetTimestamp();
var r = await evaluator.EvaluateAsync(key, n % 2 == 0 ? runaway : sibling, Scripts.Input("x", n), Ct);
samples.Add(Stopwatch.GetElapsedTime(t0).TotalMilliseconds);
r.Value.ShouldBe(n % 2 == 0 ? n * 2 : n + 100);
}
samples.Sort();
return samples;
}
List<double> reference, shadowed;
using (var plain = Calc(null)) reference = await Measure(plain);
using (var mirrored = Calc(shadow)) shadowed = await Measure(mirrored);
await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30));
var stats = shadow.Stats();
var load = LoadAverage();
var ceiling = TimeSpan.FromMilliseconds(250).TotalMilliseconds;
var report = new StringBuilder()
.AppendLine($"### Shadow published-path latency ({Env()}; load {load})").AppendLine()
.AppendLine("| Evaluator (warmed) | p50 ms | p99 ms | max ms |")
.AppendLine("|---|---:|---:|---:|")
.AppendLine($"| InProcess (no shadow) | {Percentile(reference, 0.50):F3} | {Percentile(reference, 0.99):F3} | {reference[^1]:F3} |")
.AppendLine($"| Shadow, runaway killed in the worker | {Percentile(shadowed, 0.50):F3} | {Percentile(shadowed, 0.99):F3} | {shadowed[^1]:F3} |")
.AppendLine().AppendLine($"shadow: compared {stats.Compared}, mismatched {stats.Mismatched}, unavailable {stats.ShadowUnavailable}, " +
$"dropped key_busy {stats.DroppedKeyBusy}, dropped behind {stats.DroppedBehind}; ceiling {ceiling:F0} ms per evaluation");
Report(report);
stats.Mismatched.ShouldBeGreaterThan(0, "precondition: the shadow really ran the runaway away (Failed [budget] vs published Ok)");
shadowed[^1].ShouldBeLessThan(ceiling, "a warmed published evaluation never waits for its shadow");
}
/// <summary>The host's 1/5/15-minute load average for the report: <c>/proc/loadavg</c> on Linux, <c>sysctl</c> on macOS.</summary>
private static string LoadAverage()
{
try
{
if (File.Exists("/proc/loadavg")) return File.ReadAllText("/proc/loadavg").Trim();
if (OperatingSystem.IsMacOS())
{
using var p = Process.Start(new ProcessStartInfo("/usr/sbin/sysctl", "-n vm.loadavg") { RedirectStandardOutput = true });
var text = p!.StandardOutput.ReadToEnd().Trim();
p.WaitForExit(2_000);
return text;
}
}
catch (Exception ex) when (ex is IOException or System.ComponentModel.Win32Exception or InvalidOperationException)
{
return "n/a (" + ex.GetType().Name + ")";
}
return "n/a";
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs
+116 -14
View File
@@ -8,6 +8,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
using ZB.MOM.WW.OtOpcUa.Driver.Calculation;
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
@@ -130,37 +131,89 @@ public sealed class ShadowModeTests
lines.ShouldAllBe(e => e.Level == Serilog.Events.LogEventLevel.Warning);
}
/// <summary>
/// The non-blocking contract of <see cref="IScriptShadow.Mirror"/>, proven causally rather than by a stopwatch
/// (lmxopcua#798: a 250 ms bound around the first evaluations measured cold Roslyn compile/JIT of the in-process
/// run, not shadow scheduling). <see cref="GatedShadow"/> sits between the REAL <see cref="ShadowScriptRunner"/>
/// and the REAL worker pool and holds the runaway's first shadow run at two test-owned barriers: before it reaches
/// the pool, and — after the worker has run it away, answered the soft timeout and been killed at its hard
/// deadline — before its outcome is handed back for comparison. Every published evaluation made while a barrier
/// is closed completed while its shadow was provably outstanding (only the test opens a barrier), so a published
/// path that waited for its shadow would hang here rather than merely run slow. Mutation check: make
/// <see cref="ShadowScriptRunner.Mirror"/> wait for the shadow run and the first held evaluation never returns.
/// A representative latency number is the opt-in <see cref="ScriptRunnerPerfTests"/> case (Category=Perf).
/// </summary>
[Fact]
public async Task A_runaway_that_runs_away_only_in_the_worker_is_killed_and_quarantined_there_and_never_reaches_a_published_value()
{
var log = new CapturingSink();
await using var pool = Scripts.Runner(ShadowOptions(TimeSpan.FromMilliseconds(400)), log.CreateLogger());
using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), log.CreateLogger());
var gate = new GatedShadow(pool);
using var shadow = new ShadowScriptRunner(gate, ShadowOptions(), log.CreateLogger());
using var shadowed = Calc(shadow, TimeSpan.FromMilliseconds(300));
const string runaway = $"if ({InWorker}) {{ while (true) {{ }} }} return (int)ctx.GetTag(\"i\").Value * 2;";
const string sibling = "return (int)ctx.GetTag(\"i\").Value + 100;";
var slowest = TimeSpan.Zero;
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
var i = 0;
for (; !pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"); i++)
// Off the test thread with a liveness bound (not a latency bound): a Mirror that waited for its held shadow would
// otherwise hang the test instead of failing it.
Task<VirtualTagEvalResult> Publish(string key, string source)
=> Task.Run(() => shadowed.EvaluateAsync(key, source, Inputs(i), Ct), Ct).WaitAsync(TimeSpan.FromSeconds(30), Ct);
async Task PublishBoth(string phase)
{
(await Publish("runaway", runaway)).ShouldBe(VirtualTagEvalResult.Ok(i * 2), $"{phase}: published from the in-process run");
(await Publish("sibling", sibling)).ShouldBe(VirtualTagEvalResult.Ok(i + 100), phase);
i++;
}
// Barrier 1: the runaway's first shadow run is dispatched and held before the pool ever sees it.
var held = gate.Hold("runaway");
await PublishBoth("first evaluation");
await held.Entered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
var droppedBefore = shadow.Stats().DroppedKeyBusy;
for (var k = 0; k < 3; k++)
{
await PublishBoth("shadow held before the worker");
held.ReachedPool.ShouldBeFalse("barrier 1 is still closed: this evaluation published while its shadow was blocked");
}
shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThanOrEqualTo(droppedBefore + 3,
"each runaway trigger behind the held shadow run was dropped — not queued, not waited for");
// Open barrier 1: the shadow run really runs away in the worker, the pool answers the soft timeout T and the
// worker is killed at T + G. Barrier 2 still holds the outcome back from the comparison.
var worker = await Scripts.Eventually(() => pool.Shards[0].Current, TimeSpan.FromSeconds(30), "the shadow worker");
held.ReleaseToPool();
var answered = await held.PoolAnswered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
answered.Status.ShouldBe(ScriptRunStatus.Failed, answered.Reason);
answered.Reason!.ShouldStartWith("script timed out after ", Case.Sensitive, "the worker ran it away and the pool answered at T");
await PublishBoth("shadow running away / being killed in the worker");
var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct);
exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.Calculation, "runaway"),
"only the runaway is blamed, never the sibling sharing its shard");
await PublishBoth("worker killed, shadow outcome still held");
shadow.Mismatches.ShouldBeEmpty("nothing about the held shadow run has been compared, and nothing published waited for it");
shadow.Stats().InFlight.ShouldBeGreaterThan(0, "the held shadow run is still outstanding");
// Open barrier 2: the comparison records the divergence; the published side never saw any of it.
held.Deliver();
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
shadow.Mismatches.ShouldHaveSingleItem().Key.ShouldBe("runaway");
// Un-gated from here: keep triggering until the shadow's second kill quarantines it in the pool.
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
while (!pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"))
{
if (DateTime.UtcNow > deadline)
throw new TimeoutException("the shadow never quarantined the runaway:" + Environment.NewLine + string.Join(Environment.NewLine, log.Rendered));
var sw = Stopwatch.StartNew();
var a = await shadowed.EvaluateAsync("runaway", runaway, Inputs(i), Ct);
var b = await shadowed.EvaluateAsync("sibling", sibling, Inputs(i), Ct);
if (sw.Elapsed > slowest) slowest = sw.Elapsed;
a.ShouldBe(VirtualTagEvalResult.Ok(i * 2), "published from the in-process run");
b.ShouldBe(VirtualTagEvalResult.Ok(i + 100));
await PublishBoth("until quarantined");
await Task.Delay(100, Ct);
}
// Still publishing correctly after the shadow quarantined it, and never quarantined on the published side.
for (var j = i; j < i + 5; j++)
(await shadowed.EvaluateAsync("runaway", runaway, Inputs(j), Ct)).ShouldBe(VirtualTagEvalResult.Ok(j * 2));
for (var j = 0; j < 5; j++) await PublishBoth("after the shadow quarantine");
shadowed.QuarantinedScripts.ShouldBeEmpty();
slowest.ShouldBeLessThan(TimeSpan.FromMilliseconds(250), "a published evaluation never waits for its shadow");
log.Rendered.ShouldContain(l => l.Contains("exceeded its hard deadline"));
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
@@ -171,6 +224,55 @@ public sealed class ShadowModeTests
shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThan(0, "triggers while its shadow run was still going were dropped, not queued");
}
/// <summary>
/// A pass-through shadow runner with test-owned barriers on one key's next run: it holds that run before it reaches
/// the inner runner, and holds the inner runner's outcome before returning it. Everything else forwards untouched.
/// </summary>
private sealed class GatedShadow(IScriptRunner inner) : IScriptRunner
{
private readonly ConcurrentDictionary<string, Held> _holds = new(StringComparer.Ordinal);
public Held Hold(string key) => _holds[key] = new Held();
public async Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
{
if (!_holds.TryRemove(invocation.Key, out var held)) return await inner.RunAsync(invocation, ct);
held.EnteredSource.TrySetResult();
await held.ToPool.Task.WaitAsync(ct);
held.ReachedPool = true;
var outcome = await inner.RunAsync(invocation, ct);
held.Answered.TrySetResult(outcome);
await held.Delivery.Task.WaitAsync(ct);
return outcome;
}
public void Reset(ScriptContextKind kind) => inner.Reset(kind);
public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => inner.QuarantinedKeys(kind);
public sealed class Held
{
internal readonly TaskCompletionSource EnteredSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource ToPool = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource<ScriptRunOutcome> Answered = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource Delivery = new(TaskCreationOptions.RunContinuationsAsynchronously);
private volatile bool _reachedPool;
/// <summary>The shadow runner dispatched the held run: it is outstanding from here on.</summary>
public Task Entered => EnteredSource.Task;
/// <summary>Whether the held run has been let through to the inner runner.</summary>
public bool ReachedPool { get => _reachedPool; internal set => _reachedPool = value; }
/// <summary>The inner runner's outcome for the held run, before it is returned to the shadow runner.</summary>
public Task<ScriptRunOutcome> PoolAnswered => Answered.Task;
public void ReleaseToPool() => ToPool.TrySetResult();
public void Deliver() => Delivery.TrySetResult();
}
}
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);