Joseph Doherty
304d1f1e6c
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.
345 lines
20 KiB
C#
345 lines
20 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Diagnostics;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Serilog;
|
|
using Shouldly;
|
|
using Xunit;
|
|
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;
|
|
|
|
/// <summary>
|
|
/// lmxopcua#780 P2 — <c>Scripting:Runner=Shadow</c> over the REAL worker: the seams publish their in-process results
|
|
/// and mirror each compiled evaluation to the worker pool. Ordinary scripts compare equal; a deliberately divergent
|
|
/// one (it reads which process it runs in) and an unsupported return type (design §14 #11) are counted as mismatches
|
|
/// for exactly those scripts; and a runaway that only runs away in the worker is killed and quarantined THERE while
|
|
/// every published value stays the in-process one. Alarms: the transition sequence is the in-process engine's, a
|
|
/// killed shadow worker never annotates or clears a condition.
|
|
/// </summary>
|
|
[Collection(ScriptWorkerCollection.Name)]
|
|
[Trait("Category", "ScriptWorker")]
|
|
public sealed class ShadowModeTests
|
|
{
|
|
/// <summary>
|
|
/// True only inside the ScriptWorker: a test-only divergence that needs no sandbox-forbidden API. The worker runs
|
|
/// with <c>InvariantGlobalization</c> (ordinal string comparison) while the test host does linguistic (ICU/NLS)
|
|
/// comparison, so a culture-sensitive <c>string.Compare("a", "B")</c> is negative in the host and positive in the
|
|
/// worker. <see cref="Culture_sensitive_comparison_really_diverges_between_host_and_worker"/> pins the premise.
|
|
/// </summary>
|
|
private const string InWorker = "(string.Compare(\"a\", \"B\") > 0)";
|
|
|
|
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
|
|
|
private static ScriptingOptions ShadowOptions(TimeSpan? hardGrace = null)
|
|
{
|
|
var options = Scripts.Options(hardGrace: hardGrace ?? TimeSpan.FromMilliseconds(500), quarantineAfterKills: 2);
|
|
options.Runner = ScriptRunnerMode.Shadow;
|
|
return options;
|
|
}
|
|
|
|
private static CalculationEvaluator Calc(IScriptRunner? runner, TimeSpan? budget = null)
|
|
=> new(NullLogger.Instance, new ScriptRootLogger(new LoggerConfiguration().CreateLogger()),
|
|
budget ?? TimeSpan.FromSeconds(5), scriptRunner: runner);
|
|
|
|
private static Dictionary<string, DataValueSnapshot> Inputs(int i)
|
|
{
|
|
var now = DateTime.UtcNow;
|
|
DataValueSnapshot S(object v) => new(v, 0u, now, now);
|
|
return new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal)
|
|
{
|
|
["s"] = S((short)(5 + i)),
|
|
["d"] = S(1.25 * i),
|
|
["str"] = S("abc" + i),
|
|
["i"] = S(i),
|
|
["t"] = S(new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc).AddSeconds(i)),
|
|
["arr"] = S(new[] { 1.5, 2.5 * i }),
|
|
};
|
|
}
|
|
|
|
private static bool Same(VirtualTagEvalResult a, VirtualTagEvalResult b)
|
|
=> a.Success == b.Success && a.Reason == b.Reason && a.StatusCode == b.StatusCode
|
|
&& System.Collections.StructuralComparisons.StructuralEqualityComparer.Equals(a.Value, b.Value)
|
|
&& a.Value?.GetType() == b.Value?.GetType();
|
|
|
|
[Fact]
|
|
public async Task Culture_sensitive_comparison_really_diverges_between_host_and_worker()
|
|
{
|
|
// Premise of InWorker: the worker is a different process whose globalization mode differs from the host's.
|
|
string.Compare("a", "B").ShouldBeLessThan(0, "the test host compares linguistically");
|
|
await using var pool = Scripts.Runner(Scripts.Options());
|
|
var r = await pool.RunAsync(Scripts.Vt("probe", $"return {InWorker} ? 2 : 1;"), Ct);
|
|
r.Status.ShouldBe(ScriptRunStatus.Ok, r.Reason);
|
|
r.Value.ShouldBe(2, "the worker (InvariantGlobalization) compares ordinally");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Ordinary_scripts_match_and_a_divergent_and_an_unsupported_one_are_counted_against_exactly_those_scripts()
|
|
{
|
|
var log = new CapturingSink();
|
|
await using var pool = Scripts.Runner(ShadowOptions());
|
|
using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), log.CreateLogger());
|
|
using var shadowed = Calc(shadow);
|
|
using var reference = Calc(null);
|
|
shadowed.SubmitsBatches.ShouldBeFalse("Shadow keeps the in-process one-at-a-time pump, not the runner path");
|
|
|
|
var scripts = new Dictionary<string, string>(StringComparer.Ordinal)
|
|
{
|
|
["short"] = "return (short)ctx.GetTag(\"s\").Value + 1;",
|
|
["double"] = "return (double)ctx.GetTag(\"d\").Value * 2.5;",
|
|
["string"] = "return \"v=\" + ctx.GetTag(\"str\").Value;",
|
|
["intArray"] = "return new[] { (int)ctx.GetTag(\"i\").Value, 2 };",
|
|
["doubleArray"] = "var a = (double[])ctx.GetTag(\"arr\").Value; return new[] { a[1], a[0] };",
|
|
["datetime"] = "return ((DateTime)ctx.GetTag(\"t\").Value).AddSeconds(1);",
|
|
["bool"] = "return (int)ctx.GetTag(\"i\").Value > 1;",
|
|
["quality"] = "return ctx.GetTag(\"i\").StatusCode;",
|
|
["throws"] = "if ((int)ctx.GetTag(\"i\").Value >= 0) throw new InvalidOperationException(\"boom \" + ctx.GetTag(\"i\").Value); return 0;",
|
|
["divergent"] = $"return {InWorker} ? 2 : 1;",
|
|
["decimal"] = "return 1.5m;",
|
|
};
|
|
|
|
for (var i = 0; i < 3; i++)
|
|
{
|
|
foreach (var (key, source) in scripts)
|
|
{
|
|
var published = await shadowed.EvaluateAsync(key, source, Inputs(i), Ct);
|
|
var expected = await reference.EvaluateAsync(key, source, Inputs(i), Ct);
|
|
Same(published, expected).ShouldBeTrue($"{key}: Shadow publishes exactly what InProcess publishes ({published} vs {expected})");
|
|
}
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(60))).ShouldBeTrue();
|
|
}
|
|
|
|
var stats = shadow.Stats();
|
|
stats.Dropped.ShouldBe(0);
|
|
stats.ShadowUnavailable.ShouldBe(0);
|
|
stats.Compared.ShouldBe(scripts.Count * 3);
|
|
stats.Mismatched.ShouldBe(6, string.Join(Environment.NewLine, shadow.Mismatches.Select(m => $"{m.Key}: {m.LastReason} — {m.LastSummary}")));
|
|
var mismatches = shadow.Mismatches.ToDictionary(m => m.Key);
|
|
mismatches.Keys.ShouldBe(["decimal", "divergent"], ignoreOrder: true);
|
|
mismatches["divergent"].LastReason.ShouldBe("value");
|
|
mismatches["divergent"].LastSummary.ShouldBe("published Ok Int32 1; shadow Ok Int32 2");
|
|
mismatches["divergent"].Count.ShouldBe(3);
|
|
mismatches["decimal"].LastReason.ShouldBe("unsupported_type");
|
|
mismatches["decimal"].LastSummary.ShouldContain("System.Decimal");
|
|
|
|
var lines = log.Events.Where(e => e.MessageTemplate.Text.StartsWith("Script shadow mismatch", StringComparison.Ordinal)).ToArray();
|
|
lines.Length.ShouldBe(2, "one line per script and reason");
|
|
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());
|
|
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 i = 0;
|
|
|
|
// 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));
|
|
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 = 0; j < 5; j++) await PublishBoth("after the shadow quarantine");
|
|
shadowed.QuarantinedScripts.ShouldBeEmpty();
|
|
log.Rendered.ShouldContain(l => l.Contains("exceeded its hard deadline"));
|
|
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
|
var record = shadow.Mismatches.ShouldHaveSingleItem("the sibling on the killed shard is a victim (not compared), never a mismatch");
|
|
record.Key.ShouldBe("runaway");
|
|
record.LastReason.ShouldBe("status");
|
|
record.LastSummary.ShouldContain("shadow Failed [budget]");
|
|
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);
|
|
|
|
private static ScriptedAlarmEngine Engine(TestUpstream up, IScriptRunner? runner)
|
|
{
|
|
var logger = new LoggerConfiguration().CreateLogger();
|
|
return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger,
|
|
scriptTimeout: TimeSpan.FromSeconds(2), scriptRunner: runner);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Alarms_in_shadow_transition_exactly_as_in_process_and_a_killed_shadow_worker_never_touches_a_condition()
|
|
{
|
|
await using var pool = Scripts.Runner(ShadowOptions());
|
|
using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), new LoggerConfiguration().CreateLogger());
|
|
var up = new TestUpstream();
|
|
var refUp = new TestUpstream();
|
|
up.Set("x", 0);
|
|
refUp.Set("x", 0);
|
|
using var engine = Engine(up, shadow);
|
|
using var reference = Engine(refUp, null);
|
|
engine.ScriptRunner.ShouldBeNull("Shadow keeps the in-process engine path");
|
|
var events = new ConcurrentQueue<ScriptedAlarmEvent>();
|
|
var refEvents = new ConcurrentQueue<ScriptedAlarmEvent>();
|
|
engine.OnEvent += (_, e) => events.Enqueue(e);
|
|
reference.OnEvent += (_, e) => refEvents.Enqueue(e);
|
|
ScriptedAlarmDefinition[] defs =
|
|
[
|
|
Alarm("high", "return (int)ctx.GetTag(\"x\").Value > 5;"),
|
|
Alarm("divergent", $"return {InWorker};"),
|
|
];
|
|
await engine.LoadAsync(defs, Ct);
|
|
await reference.LoadAsync(defs, Ct);
|
|
|
|
async Task Push(int x)
|
|
{
|
|
up.Push("x", x);
|
|
refUp.Push("x", x);
|
|
await Scripts.Eventually(() => events.Count == refEvents.Count, TimeSpan.FromSeconds(10), "both engines' emissions");
|
|
await Task.Delay(100, Ct);
|
|
}
|
|
|
|
await Push(10);
|
|
await Push(2);
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
|
|
|
// kill -9 the shadow's worker: the published engine neither annotates nor clears anything because of it.
|
|
var worker = pool.Shards[0].Current.ShouldNotBeNull();
|
|
Process.GetProcessById(worker.Pid).Kill();
|
|
await Push(12);
|
|
await Push(0);
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
|
await Push(9);
|
|
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
|
|
|
events.Select(e => (e.AlarmId, e.Emission, e.WorstInputStatusCode))
|
|
.ShouldBe(refEvents.Select(e => (e.AlarmId, e.Emission, e.WorstInputStatusCode)), "the in-process engine's sequence, exactly");
|
|
events.Count(e => e is { AlarmId: "high", Emission: EmissionKind.Activated }).ShouldBe(3);
|
|
events.ShouldNotContain(e => e.WorstInputStatusCode != 0u, "no shadow-side failure annotates a published condition");
|
|
engine.GetState("divergent")!.Active.ShouldBe(AlarmActiveState.Inactive, "the in-process predicate (false) drives it");
|
|
|
|
var record = shadow.Mismatches.ShouldHaveSingleItem();
|
|
record.Key.ShouldBe("divergent");
|
|
record.Kind.ShouldBe(ScriptContextKind.AlarmPredicate);
|
|
record.LastSummary.ShouldBe("published Ok Boolean False; shadow Ok Boolean True");
|
|
shadow.Stats().Matched.ShouldBeGreaterThan(0);
|
|
}
|
|
}
|