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;
///
/// lmxopcua#780 P2 — Scripting:Runner=Shadow 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.
///
[Collection(ScriptWorkerCollection.Name)]
[Trait("Category", "ScriptWorker")]
public sealed class ShadowModeTests
{
///
/// True only inside the ScriptWorker: a test-only divergence that needs no sandbox-forbidden API. The worker runs
/// with InvariantGlobalization (ordinal string comparison) while the test host does linguistic (ICU/NLS)
/// comparison, so a culture-sensitive string.Compare("a", "B") is negative in the host and positive in the
/// worker. pins the premise.
///
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, CapturingSink? scriptLog = null)
=> new(NullLogger.Instance,
new ScriptRootLogger(scriptLog is null
? new LoggerConfiguration().CreateLogger()
: new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(scriptLog).CreateLogger()),
budget ?? TimeSpan.FromSeconds(5), scriptRunner: runner);
private static Dictionary Inputs(int i)
{
var now = DateTime.UtcNow;
DataValueSnapshot S(object v) => new(v, 0u, now, now);
return new Dictionary(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(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);
}
///
/// The non-blocking contract of , 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). sits between the REAL
/// 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
/// wait for the shadow run and the first held evaluation never returns.
/// A representative latency number is the opt-in case (Category=Perf).
///
[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();
var scriptLog = 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, RunawayRunBudget, scriptLog);
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;
var testClock = Stopwatch.StartNew();
// 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 Publish(string key, string source)
=> Task.Run(() => shadowed.EvaluateAsync(key, source, Inputs(i), Ct), Ct).WaitAsync(TimeSpan.FromSeconds(30), Ct);
// lmxopcua#811: a wrong published value must say what it was and what the host was doing — the one loaded
// failure seen so far was unattributable because only "should be Ok(…)" survived a truncated log.
async Task Expect(string phase, string key, string source, int expected)
{
var before = PublishProbe.Take(shadowed, shadow);
var sw = Stopwatch.StartNew();
var published = await Publish(key, source);
sw.Stop();
if (published == VirtualTagEvalResult.Ok(expected)) return;
var after = PublishProbe.Take(shadowed, shadow);
published.ShouldBe(VirtualTagEvalResult.Ok(expected),
$"{phase}: {key} (iteration {i}) published the wrong value — " +
$"Success={published.Success}, Value={published.Value ?? "null"} ({published.Value?.GetType().Name ?? "-"}), " +
$"Reason={published.Reason ?? "null"}, StatusCode={(published.StatusCode is { } sc ? $"0x{sc:X8}" : "null")}, " +
$"RetryWhen={(published.RetryWhen is null ? "null" : "set")}; publish took {sw.Elapsed.TotalMilliseconds:F0} ms " +
$"against a {RunawayRunBudget.TotalMilliseconds:F0} ms run budget, at t+{testClock.Elapsed.TotalSeconds:F1} s; " +
$"quarantined={(shadowed.RunGuard.IsQuarantined(key, out var why) ? why : "no")}" + Environment.NewLine +
$" before: {before}" + Environment.NewLine +
$" after: {after}" + Environment.NewLine +
" script log (last 20):" + Lines(scriptLog) + Environment.NewLine +
" runner log (last 20):" + Lines(log));
}
static string Lines(CapturingSink sink)
=> sink.Events.IsEmpty
? " (none)"
: Environment.NewLine + " " + string.Join(Environment.NewLine + " ",
sink.Events.TakeLast(20).Select(e => $"{e.Timestamp:HH:mm:ss.fff} {e.Level} {e.RenderMessage()}"));
async Task PublishBoth(string phase)
{
await Expect($"{phase}: published from the in-process run", "runaway", runaway, i * 2);
await Expect(phase, "sibling", sibling, i + 100);
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");
}
///
/// The in-process run budget of the runaway test's published path. It is also the soft deadline the shadow worker
/// answers the runaway at (the mirrored invocation carries the seam's budget), so it sets how long the shadow runs
/// away before it is killed.
/// lmxopcua#811: it was 300 ms, and under heavy CPU load (32 yes processes, the test at nice 20, load
/// average 300+) 3 of 8 runs failed because the SIBLING's shadow — a one-line script — missed that soft deadline
/// in the worker ("shadow Failed [budget] script timed out after 0.3s"), which the comparison rightly records as a
/// sibling mismatch. The out-of-process soft deadline is armed at submit, so it also covers the pipe round trip
/// and the starved worker being scheduled at all. The published values were never wrong. 2 s is the production
/// default budget; the runaway still runs away (and is killed at T + HardGrace) whatever T is, so nothing the test
/// proves depends on T being short — only its runtime does (two kills, ~5 s).
///
private static readonly TimeSpan RunawayRunBudget = TimeSpan.FromSeconds(2);
/// Host-side state around one published evaluation, for a self-diagnosing failure message (lmxopcua#811).
private readonly record struct PublishProbe(
int GuardInFlight, long GuardStarted, long GuardRefused, long GuardQuarantined, int PoolThreads, long PoolPending,
int Gen0, int Gen1, int Gen2, double GcPauseMs, ScriptShadowStats Shadow)
{
public static PublishProbe Take(CalculationEvaluator evaluator, ShadowScriptRunner shadow) => new(
evaluator.RunGuard.InFlightCount, evaluator.RunGuard.StartedRunCount, evaluator.RunGuard.RefusedWhileRunningCount,
evaluator.RunGuard.QuarantinedCount, ThreadPool.ThreadCount, ThreadPool.PendingWorkItemCount,
GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2), GC.GetTotalPauseDuration().TotalMilliseconds,
shadow.Stats());
public override string ToString() =>
$"guard inFlight={GuardInFlight} started={GuardStarted} refused={GuardRefused} quarantined={GuardQuarantined}; " +
$"pool threads={PoolThreads} pending={PoolPending}; gc gen0/1/2={Gen0}/{Gen1}/{Gen2} totalPause={GcPauseMs:F1} ms; " +
$"shadow {Shadow}";
}
///
/// 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.
///
private sealed class GatedShadow(IScriptRunner inner) : IScriptRunner
{
private readonly ConcurrentDictionary _holds = new(StringComparer.Ordinal);
public Held Hold(string key) => _holds[key] = new Held();
public async Task 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 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 Answered = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource Delivery = new(TaskCreationOptions.RunContinuationsAsynchronously);
private volatile bool _reachedPool;
/// The shadow runner dispatched the held run: it is outstanding from here on.
public Task Entered => EnteredSource.Task;
/// Whether the held run has been let through to the inner runner.
public bool ReachedPool { get => _reachedPool; internal set => _reachedPool = value; }
/// The inner runner's outcome for the held run, before it is returned to the shadow runner.
public Task 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();
var refEvents = new ConcurrentQueue();
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);
}
}