using System.Collections.Concurrent; using System.Diagnostics; using System.Diagnostics.Metrics; using Serilog; using Serilog.Core; using Serilog.Events; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Engines; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; /// /// lmxopcua#780 P2 — , the comparison half of Scripting:Runner=Shadow, against /// fake shadow runners: the comparison table, per-script mismatch accounting and its rate-limited log, and the /// guarantee that nothing on the shadow side — a runner that hangs, blocks its caller, or throws — reaches the /// evaluation that published, plus the drop accounting that keeps the shadow bounded. /// public sealed class ShadowScriptRunnerTests { private static readonly IReadOnlyDictionary NoInputs = new Dictionary(StringComparer.Ordinal); private sealed class FakeShadow(Func> run) : IScriptRunner { public ConcurrentQueue Invocations { get; } = new(); public ConcurrentQueue Resets { get; } = new(); public Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default) { Invocations.Enqueue(invocation); return run(invocation, ct); } public void Reset(ScriptContextKind kind) => Resets.Enqueue(kind); public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => []; } private sealed class Capture : ILogEventSink { public ConcurrentQueue Events { get; } = new(); public void Emit(LogEvent logEvent) => Events.Enqueue(logEvent); public ILogger Logger() => new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(this).CreateLogger(); public string[] Mismatches => Events.Where(e => e.MessageTemplate.Text.StartsWith("Script shadow mismatch", StringComparison.Ordinal)) .Select(e => e.RenderMessage()).ToArray(); } private static ScriptInvocation Vt(string key, IReadOnlyDictionary? inputs = null, string source = "return 1;") => new(key, ScriptContextKind.VirtualTag, source, inputs ?? NoInputs, TimeSpan.FromSeconds(2)); private static ScriptingOptions Options(int maxInFlight = 256, int logsPerMinute = 20) => new() { Shadow = new ScriptShadowOptions { MaxInFlight = maxInFlight, MismatchLogsPerMinute = logsPerMinute } }; private static string NewKey() => "shadow-" + Guid.NewGuid().ToString("N")[..10]; public static TheoryData Table => new() { { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Ok(null), ScriptRunOutcome.Ok(null), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Ok(double.NaN), ScriptRunOutcome.Ok(double.NaN), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(43), ScriptShadowVerdict.Mismatch, "value" }, { ScriptRunOutcome.Ok(new[] { 1, 2 }), ScriptRunOutcome.Ok(new[] { 1, 3 }), ScriptShadowVerdict.Mismatch, "value" }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Ok(null), ScriptShadowVerdict.Mismatch, "value" }, { ScriptRunOutcome.Ok((short)42), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Mismatch, "type" }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed("script threw: boom"), ScriptShadowVerdict.Mismatch, "status" }, { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Ok(42), ScriptShadowVerdict.Mismatch, "status" }, { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Failed("script threw: boom"), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Failed("script threw: boom"), ScriptRunOutcome.Failed("script threw: bang"), ScriptShadowVerdict.Mismatch, "exception" }, { ScriptRunOutcome.Failed("compile error: CS1002"), ScriptRunOutcome.Failed("compile error: CS1002 (worker)"), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Failed("compile error: CS1002"), ScriptRunOutcome.Failed("script threw: x"), ScriptShadowVerdict.Mismatch, "failure_kind" }, // The same runaway reaches "budget" by different routes in the two runners: a timeout here, a kill-strike quarantine there. { ScriptRunOutcome.Failed("script timed out after 2.0s"), ScriptRunOutcome.Failed("script quarantined: killed twice"), ScriptShadowVerdict.Match, null }, { ScriptRunOutcome.Failed("previous evaluation still running"), ScriptRunOutcome.Failed("script timed out after 2.0s"), ScriptShadowVerdict.Match, null }, // Design §14 #11: what the protocol cannot carry is surfaced in shadow first. { ScriptRunOutcome.Ok(1.5m), ScriptRunOutcome.Failed("unsupported type: script returned a value of unsupported type System.Decimal"), ScriptShadowVerdict.Mismatch, "unsupported_type" }, { ScriptRunOutcome.Failed("script threw: x"), ScriptRunOutcome.Failed("unsupported type: input 'a' has a value of unsupported type System.Guid"), ScriptShadowVerdict.Mismatch, "unsupported_type" }, // The runner's fault, not the script's: not compared. { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), ScriptShadowVerdict.ShadowUnavailable, null }, { ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed(ScriptRunOutcome.WorkerBusyReason), ScriptShadowVerdict.ShadowUnavailable, null }, }; [Theory] [MemberData(nameof(Table))] public void Compare_classifies_each_pair(ScriptRunOutcome published, ScriptRunOutcome shadow, ScriptShadowVerdict verdict, string? reason) { var comparison = ShadowScriptRunner.Compare(published, shadow); comparison.Verdict.ShouldBe(verdict); comparison.Reason.ShouldBe(reason); } [Fact] public async Task A_divergent_script_is_counted_per_script_in_metrics_and_logged_once_with_a_diff_summary() { var divergent = NewKey(); var healthy = NewKey(); var shadow = new FakeShadow((inv, _) => Task.FromResult(inv.Key == divergent ? ScriptRunOutcome.Ok(2) : ScriptRunOutcome.Ok(1))); var log = new Capture(); using var runner = new ShadowScriptRunner(shadow, Options(), log.Logger()); var measured = new ConcurrentQueue<(long Value, string? Script, string? Reason, string? Kind)>(); using var listener = new MeterListener(); listener.InstrumentPublished = (instrument, l) => { if (instrument.Name == "otopcua.script.shadow.mismatch") l.EnableMeasurementEvents(instrument); }; listener.SetMeasurementEventCallback((_, value, tags, _) => { string? script = null, reason = null, kind = null; foreach (var tag in tags) { if (tag.Key == "script") script = tag.Value as string; if (tag.Key == "reason") reason = tag.Value as string; if (tag.Key == "kind") kind = tag.Value as string; } if (script == divergent || script == healthy) measured.Enqueue((value, script, reason, kind)); }); listener.Start(); for (var i = 0; i < 5; i++) { runner.Mirror(Vt(divergent), ScriptRunOutcome.Ok(1)); runner.Mirror(Vt(healthy), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); } var stats = runner.Stats(); stats.Compared.ShouldBe(10); stats.Matched.ShouldBe(5); stats.Mismatched.ShouldBe(5); stats.Dropped.ShouldBe(0); var record = runner.Mismatches.ShouldHaveSingleItem(); record.Key.ShouldBe(divergent); record.Count.ShouldBe(5); record.LastReason.ShouldBe("value"); record.LastSummary.ShouldBe("published Ok Int32 1; shadow Ok Int32 2"); measured.Count.ShouldBe(5); measured.ShouldAllBe(m => m.Script == divergent && m.Reason == "value" && m.Kind == "virtual_tag" && m.Value == 1); var line = log.Mismatches.ShouldHaveSingleItem("once per script and reason, not once per evaluation"); line.ShouldContain(divergent); line.ShouldContain("published Ok Int32 1; shadow Ok Int32 2"); } [Fact] public async Task Mismatch_lines_are_rate_limited_per_minute_and_the_next_line_reports_what_was_suppressed() { var now = new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc); var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(2))); var log = new Capture(); using var runner = new ShadowScriptRunner(shadow, Options(logsPerMinute: 2), log.Logger(), clock: () => now); for (var i = 0; i < 5; i++) runner.Mirror(Vt(NewKey()), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); log.Mismatches.Length.ShouldBe(2); runner.Stats().Mismatched.ShouldBe(5, "every mismatch is counted even when its line is not written"); now = now.AddMinutes(1); runner.Mirror(Vt(NewKey()), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); log.Mismatches.Length.ShouldBe(3); log.Mismatches[^1].ShouldContain("3 earlier mismatch line(s) suppressed"); } [Fact] public async Task A_reset_re_arms_the_once_per_script_line_and_resets_the_shadow_runner() { var key = NewKey(); var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(2))); var log = new Capture(); using var runner = new ShadowScriptRunner(shadow, Options(), log.Logger()); runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); log.Mismatches.Length.ShouldBe(1); runner.Reset(ScriptContextKind.Calculation); runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); log.Mismatches.Length.ShouldBe(1, "another seam's deploy does not re-arm a virtual tag's line"); runner.Reset(ScriptContextKind.VirtualTag); runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); log.Mismatches.Length.ShouldBe(2); shadow.Resets.ShouldBe([ScriptContextKind.Calculation, ScriptContextKind.VirtualTag]); } [Fact] public async Task A_shadow_runner_that_hangs_blocks_or_throws_never_reaches_the_caller_and_is_never_a_mismatch() { var release = new ManualResetEventSlim(false); var hang = NewKey(); var block = NewKey(); var boom = NewKey(); var shadow = new FakeShadow((inv, _) => { if (inv.Key == hang) return new TaskCompletionSource().Task; // never completes if (inv.Key == block) release.Wait(TimeSpan.FromSeconds(2)); // blocks the thread that called RunAsync (bounded, so a regression fails instead of hanging) if (inv.Key == boom) throw new InvalidOperationException("runner bug"); return Task.FromResult(ScriptRunOutcome.Ok(1)); }); using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); var sw = Stopwatch.StartNew(); runner.Mirror(Vt(hang), ScriptRunOutcome.Ok(1)); runner.Mirror(Vt(block), ScriptRunOutcome.Ok(1)); runner.Mirror(Vt(boom), ScriptRunOutcome.Ok(1)); sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500), "Mirror never runs the shadow on the caller's thread"); // Each key's run is outstanding (or done): a second mirror of the hung/blocked keys is dropped, never queued. await Task.Delay(100, TestContext.Current.CancellationToken); runner.Mirror(Vt(hang), ScriptRunOutcome.Ok(1)); runner.Mirror(Vt(block), ScriptRunOutcome.Ok(1)); runner.Stats().DroppedKeyBusy.ShouldBe(2); release.Set(); await Task.Delay(200, TestContext.Current.CancellationToken); var stats = runner.Stats(); stats.Mismatched.ShouldBe(0); stats.ShadowUnavailable.ShouldBe(1, "a runner that throws is the runner's fault, not a divergence"); stats.Matched.ShouldBe(1, "the blocked one completed once released"); stats.InFlight.ShouldBe(1, "the hung one is still outstanding — bounded at one per key"); } [Fact] public async Task Evaluations_beyond_MaxInFlight_are_dropped_and_counted_and_the_shadow_resumes_once_it_catches_up() { var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var shadow = new FakeShadow((_, _) => gate.Task); using var runner = new ShadowScriptRunner(shadow, Options(maxInFlight: 3), new LoggerConfiguration().CreateLogger()); var measured = new ConcurrentQueue(); using var listener = new MeterListener(); listener.InstrumentPublished = (instrument, l) => { if (instrument.Name == "otopcua.script.shadow.dropped") l.EnableMeasurementEvents(instrument); }; var keys = Enumerable.Range(0, 10).Select(_ => NewKey()).ToArray(); listener.SetMeasurementEventCallback((_, _, tags, _) => { foreach (var tag in tags) if (tag.Key == "reason") measured.Enqueue(tag.Value as string); }); listener.Start(); foreach (var key in keys) runner.Mirror(Vt(key), ScriptRunOutcome.Ok(1)); await Scripts.Eventually(() => shadow.Invocations.Count == 3, "three shadow runs started"); var stats = runner.Stats(); stats.InFlight.ShouldBe(3); stats.DroppedBehind.ShouldBe(7); shadow.Invocations.Count.ShouldBe(3, "a dropped evaluation never reaches the shadow runner"); gate.SetResult(ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); runner.Mirror(Vt(keys[9]), ScriptRunOutcome.Ok(1)); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); stats = runner.Stats(); stats.Matched.ShouldBe(4); stats.DroppedBehind.ShouldBe(7); measured.Count(r => r == "behind").ShouldBeGreaterThanOrEqualTo(7); } [Fact] public async Task Mirror_copies_the_inputs_before_it_returns_and_never_forwards_the_script_logger() { var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var seen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var shadow = new FakeShadow((inv, _) => { seen.TrySetResult(inv); return gate.Task; }); using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); var cache = new Dictionary(StringComparer.Ordinal) { ["a"] = new(1, 0u, DateTime.UtcNow, DateTime.UtcNow), }; var scriptLog = new LoggerConfiguration().CreateLogger(); runner.Mirror(Vt(NewKey(), cache) with { ScriptLogger = scriptLog }, ScriptRunOutcome.Ok(1)); cache["a"] = new DataValueSnapshot(99, 0u, DateTime.UtcNow, DateTime.UtcNow); // the alarm engine refills in place var invocation = await seen.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); gate.SetResult(ScriptRunOutcome.Ok(1)); invocation.Inputs["a"].Value.ShouldBe(1); invocation.Inputs.ShouldNotBeSameAs(cache); invocation.ScriptLogger.ShouldBeSameAs(Logger.None, "the in-process run already wrote the script's log lines"); } [Fact] public async Task As_a_runner_it_publishes_the_in_process_result_and_mirrors_it() { var shadow = new FakeShadow((_, _) => Task.FromResult(ScriptRunOutcome.Ok(-1))); using var runner = new ShadowScriptRunner(shadow, Options(), new LoggerConfiguration().CreateLogger()); var inputs = new Dictionary(StringComparer.Ordinal) { ["a"] = new(20, 0u, DateTime.UtcNow, DateTime.UtcNow), }; var outcome = await runner.RunAsync(Vt(NewKey(), inputs, "return (int)ctx.GetTag(\"a\").Value + 1;"), TestContext.Current.CancellationToken); outcome.Status.ShouldBe(ScriptRunStatus.Ok); outcome.Value.ShouldBe(21, "the in-process result, not the shadow's -1"); (await runner.WaitIdleAsync(TimeSpan.FromSeconds(10))).ShouldBeTrue(); runner.Stats().Mismatched.ShouldBe(1); } [Theory] [InlineData("evaluator disposed")] [InlineData(ScriptEvalFailureReasons.EvaluatorDisposedDuringEvaluation)] [InlineData("evaluation cancelled (driver shutting down)")] public void Host_lifecycle_failures_are_not_mirrored(string reason) => VirtualTagEvalResult.Failure(reason).ToMirroredOutcome().ShouldBeNull(); [Fact] public void Script_results_map_onto_runner_outcomes() { VirtualTagEvalResult.Ok(5).ToMirroredOutcome()!.ShouldBe(ScriptRunOutcome.Ok(5)); VirtualTagEvalResult.Failure("script threw: x").ToMirroredOutcome()!.ShouldBe(ScriptRunOutcome.Failed("script threw: x")); VirtualTagEvalResult.Unavailable("script worker restarted").ToMirroredOutcome()!.Status.ShouldBe(ScriptRunStatus.Unavailable); } private static class Scripts { public static async Task Eventually(Func probe, string what) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); while (DateTime.UtcNow < deadline) { if (probe()) return; await Task.Delay(10); } throw new TimeoutException($"timed out waiting for {what}"); } } }