lmxopcua/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ShadowScriptRunnerTests.cs

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;

/// <summary>
///     lmxopcua#780 P2 — <see cref="ShadowScriptRunner"/>, the comparison half of <c>Scripting:Runner=Shadow</c>, 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.
/// </summary>
public sealed class ShadowScriptRunnerTests
{
    private static readonly IReadOnlyDictionary<string, DataValueSnapshot> NoInputs =
        new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);

    private sealed class FakeShadow(Func<ScriptInvocation, CancellationToken, Task<ScriptRunOutcome>> run) : IScriptRunner
    {
        public ConcurrentQueue<ScriptInvocation> Invocations { get; } = new();
        public ConcurrentQueue<ScriptContextKind> Resets { get; } = new();

        public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
        {
            Invocations.Enqueue(invocation);
            return run(invocation, ct);
        }

        public void Reset(ScriptContextKind kind) => Resets.Enqueue(kind);

        public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => [];
    }

    private sealed class Capture : ILogEventSink
    {
        public ConcurrentQueue<LogEvent> 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<string, DataValueSnapshot>? 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<ScriptRunOutcome, ScriptRunOutcome, ScriptShadowVerdict, string?> 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<long>((_, 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<ScriptRunOutcome>().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<ScriptRunOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
        var shadow = new FakeShadow((_, _) => gate.Task);
        using var runner = new ShadowScriptRunner(shadow, Options(maxInFlight: 3), new LoggerConfiguration().CreateLogger());

        var measured = new ConcurrentQueue<string?>();
        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<long>((_, _, 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<ScriptRunOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
        var seen = new TaskCompletionSource<ScriptInvocation>(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<string, DataValueSnapshot>(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<string, DataValueSnapshot>(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<bool> 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}");
        }
    }
}