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/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);