lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/BatchSubmitTests.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;

namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;

/// <summary>
///     lmxopcua#780 batch submit (design §4.5) against the real worker: <see cref="IScriptRunner.RunBatch"/> sends one frame
///     per worker and answers per item, and every item keeps its own deadlines and blame — in particular an item that has to
///     wait in the worker behind a runaway sibling is neither timed out, nor struck, nor blamed for the kill.
/// </summary>
[Collection(ScriptWorkerCollection.Name)]
[Trait("Category", "ScriptWorker")]
public sealed class BatchSubmitTests
{
    /// <summary>Runs away when input <c>spin</c> is true, otherwise returns <c>n + 1</c>.</summary>
    private const string MaybeRunaway = "if ((bool)ctx.GetTag(\"spin\").Value) { while (true) { } } return (int)ctx.GetTag(\"n\").Value + 1;";

    /// <summary>Spins for input <c>ms</c> milliseconds, then returns it.</summary>
    private const string Spin = "var ms = (int)ctx.GetTag(\"ms\").Value; var end = ctx.Now.AddMilliseconds(ms); while (ctx.Now < end) { } return ms;";

    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    private static IReadOnlyDictionary<string, DataValueSnapshot> Inputs(params (string Path, object Value)[] values)
    {
        var map = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
        foreach (var (path, value) in values) map[path] = new DataValueSnapshot(value, 0, DateTime.UtcNow, DateTime.UtcNow);
        return map;
    }

    private static ScriptInvocation Item(string key, bool spin, int n, TimeSpan? timeout = null)
        => Scripts.Vt(key, MaybeRunaway, timeout, Inputs(("spin", spin), ("n", n)));

    private static OutOfProcessScriptRunner Runner(ScriptingOptions options, int evalThreads)
        => new(options, new LoggerConfiguration().CreateLogger(), null,
            heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(60), evalThreads: evalThreads);

    [Fact]
    public async Task A_batch_of_loaded_scripts_goes_out_as_one_frame_and_every_item_gets_its_own_result()
    {
        await using var runner = Runner(Scripts.Options(), evalThreads: 4);
        // First evaluations load (compile) each key; those take the full path one by one.
        var warm = await Task.WhenAll(runner.RunBatch(Enumerable.Range(0, 40).Select(i => Item($"k{i}", false, i)).ToArray(), Ct));
        warm.ShouldAllBe(o => o.Status == ScriptRunStatus.Ok);

        var frames = new ConcurrentQueue<int>();
        runner.BatchSubmittedForTest = frames.Enqueue;
        var outcomes = await Task.WhenAll(runner.RunBatch(Enumerable.Range(0, 40).Select(i => Item($"k{i}", false, i * 10)).ToArray(), Ct));

        frames.ToArray().ShouldBe([40], "every loaded item rides one EvalBatch frame");
        for (var i = 0; i < 40; i++)
        {
            outcomes[i].Status.ShouldBe(ScriptRunStatus.Ok, outcomes[i].Reason);
            outcomes[i].Value.ShouldBe(i * 10 + 1, $"item {i} got its own result");
        }
    }

    [Fact]
    public async Task A_batch_answers_exactly_what_the_same_evaluations_answer_one_by_one()
    {
        // Parity: Ok, a throw, a compile error, a duplicate key (refused) and a fast item, batched vs RunAsync.
        ScriptInvocation[] Items() =>
        [
            Scripts.Vt("ok", "return (int)ctx.GetTag(\"n\").Value * 2;", inputs: Inputs(("n", 21))),
            Scripts.Vt("throws", "throw new System.InvalidOperationException(\"boom\");"),
            Scripts.Vt("broken", "return undefinedThing;"),
            Scripts.Vt("fast", "return 5;"),
        ];

        await using var runner = Runner(Scripts.Options(), evalThreads: 2);
        var single = new List<ScriptRunOutcome>();
        foreach (var item in Items()) single.Add(await runner.RunAsync(item, Ct));
        var batched = await Task.WhenAll(runner.RunBatch(Items(), Ct));

        batched.Select(o => (o.Status, o.Value, o.Reason)).ShouldBe(single.Select(o => (o.Status, o.Value, o.Reason)));
        single[0].Value.ShouldBe(42);
        single[1].Reason!.ShouldStartWith("script threw: ");
        single[2].Reason!.ShouldStartWith("compile error: ");

        // In process, the default RunBatch is RunAsync per item: same answers.
        using var inProcess = new InProcessScriptRunner();
        var inProcessBatched = await Task.WhenAll(((IScriptRunner)inProcess).RunBatch(Items(), Ct));
        inProcessBatched.Select(o => (o.Status, o.Value)).ShouldBe(single.Select(o => (o.Status, o.Value)));

        // A key twice in one batch: the second is refused, exactly like two concurrent RunAsync calls.
        var twice = await Task.WhenAll(runner.RunBatch([Scripts.Vt("dup", Spin, inputs: Inputs(("ms", 200))), Scripts.Vt("dup", Spin, inputs: Inputs(("ms", 1)))], Ct));
        twice.Count(o => o.Reason == ScriptRunGuard.PreviousRunStillRunningReason).ShouldBe(1);
    }

    /// <summary>
    ///     The blame rule (design §5.3.1) under batching: with ONE evaluation thread, a runaway at the head of a batch holds
    ///     the thread and its siblings wait in the worker. The siblings are the kill's victims — Unavailable "worker
    ///     restarted", no timeout, no strike — and only the runaway is killed and quarantined. Mutation check: with the
    ///     host ignoring the worker's deferral notice, the siblings' clocks run from submit, they time out at T with the
    ///     runaway, and at T + G any of the four can be the one blamed.
    /// </summary>
    [Fact]
    public async Task A_runaway_in_a_batch_is_blamed_alone_and_siblings_queued_behind_it_are_not_timed_out_or_struck()
    {
        var softT = TimeSpan.FromMilliseconds(400);
        var grace = TimeSpan.FromMilliseconds(500);
        await using var runner = Runner(Scripts.Options(hardGrace: grace, quarantineAfterKills: 1), evalThreads: 1);
        var keys = new[] { "runaway", "sib1", "sib2", "sib3" };
        foreach (var key in keys) (await runner.RunAsync(Item(key, false, 1, softT), Ct)).Value.ShouldBe(2);
        var worker = runner.Shards[0].Current.ShouldNotBeNull();

        var frames = new ConcurrentQueue<int>();
        runner.BatchSubmittedForTest = frames.Enqueue;
        var sw = Stopwatch.StartNew();
        var tasks = runner.RunBatch([Item("runaway", true, 0, softT), Item("sib1", false, 10, softT), Item("sib2", false, 20, softT), Item("sib3", false, 30, softT)], Ct);
        frames.ToArray().ShouldBe([4]);

        var runaway = await tasks[0];
        runaway.Status.ShouldBe(ScriptRunStatus.Failed);
        runaway.Reason!.ShouldStartWith("script timed out after ");
        // Nothing has answered the queued siblings at the runaway's soft deadline.
        tasks.Skip(1).ShouldAllBe(t => !t.IsCompleted, "a run waiting behind a busy thread is not timed out at its submit-relative T");

        var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
        exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
        exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, "runaway"));
        sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50));

        var siblings = await Task.WhenAll(tasks.Skip(1));
        foreach (var sibling in siblings)
        {
            sibling.Status.ShouldBe(ScriptRunStatus.Unavailable, "a victim of another key's kill, not a script failure");
            sibling.Reason.ShouldBe(ScriptRunOutcome.WorkerRestartedReason);
        }
        await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count > 0, TimeSpan.FromSeconds(5), "the kill quarantine");
        runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["runaway"], "the siblings took no strike");

        // The siblings run again on the restarted worker, with their own inputs.
        var again = await Task.WhenAll(runner.RunBatch([Item("sib1", false, 10, softT), Item("sib2", false, 20, softT), Item("sib3", false, 30, softT)], Ct));
        again.Select(o => o.Value).ShouldBe([11, 21, 31]);
    }

    /// <summary>
    ///     The positive control for the deferral: a run that waited behind a sibling still gets its deadlines — the full T,
    ///     then G — counted from when it STARTED. Mutation check: a start notice that does not re-arm the clock leaves the
    ///     deferred runaway with no deadline at all; it is never answered or killed.
    /// </summary>
    [Fact]
    public async Task A_deferred_run_is_timed_and_killed_from_when_it_starts()
    {
        var softT = TimeSpan.FromMilliseconds(400);
        var grace = TimeSpan.FromMilliseconds(500);
        var head = TimeSpan.FromMilliseconds(300);
        await using var runner = Runner(Scripts.Options(hardGrace: grace, quarantineAfterKills: 5), evalThreads: 1);
        (await runner.RunAsync(Scripts.Vt("head", Spin, softT, Inputs(("ms", 1))), Ct)).Value.ShouldBe(1);
        (await runner.RunAsync(Item("tail", false, 1, softT), Ct)).Value.ShouldBe(2);
        var worker = runner.Shards[0].Current.ShouldNotBeNull();

        var sw = Stopwatch.StartNew();
        var tasks = runner.RunBatch([Scripts.Vt("head", Spin, softT, Inputs(("ms", (int)head.TotalMilliseconds))), Item("tail", true, 0, softT)], Ct);

        (await tasks[0]).Value.ShouldBe((int)head.TotalMilliseconds, "the head ran inside its budget");
        var tail = await tasks[1];
        var answered = sw.Elapsed;
        tail.Reason!.ShouldStartWith("script timed out after ");
        answered.ShouldBeGreaterThanOrEqualTo(head + softT - TimeSpan.FromMilliseconds(50), "T counts from the run's start, after the head");

        var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
        exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, "tail"));
        sw.Elapsed.ShouldBeGreaterThanOrEqualTo(head + softT + grace - TimeSpan.FromMilliseconds(50), "G counts from the run's start too");
    }
}