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

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

namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests;

/// <summary>
///     lmxopcua#780 (design §4.5) — the virtual-tag seam's opportunistic batching: an evaluation on an idle seam is
///     submitted at once by its own caller (no timer, no added latency), and evaluations submitted while a submission
///     is in progress go to the runner together, in arrival order, each caller getting its own outcome.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ScriptBatchCoalescerTests
{
    private static readonly IReadOnlyDictionary<string, DataValueSnapshot> NoInputs = new Dictionary<string, DataValueSnapshot>();

    private static ScriptInvocation Vt(string key) => new(key, ScriptContextKind.VirtualTag, "return 1;", NoInputs, TimeSpan.FromSeconds(1));

    /// <summary>Answers each key with itself; the first <see cref="RunBatch"/> blocks until released.</summary>
    private sealed class BlockingRunner : IScriptRunner
    {
        private int _calls;

        public ManualResetEventSlim Entered { get; } = new();

        public ManualResetEventSlim Release { get; } = new();

        public ConcurrentQueue<string[]> Batches { get; } = new();

        public bool BlockFirst { get; init; } = true;

        public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
            => throw new InvalidOperationException("the coalescer must submit through RunBatch");

        public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
        {
            Batches.Enqueue(invocations.Select(i => i.Key).ToArray());
            if (BlockFirst && Interlocked.Increment(ref _calls) == 1)
            {
                Entered.Set();
                Release.Wait(TimeSpan.FromSeconds(10)).ShouldBeTrue();
            }
            return invocations.Select(i => Task.FromResult(ScriptRunOutcome.Ok(i.Key))).ToArray();
        }

        public void Reset(ScriptContextKind kind) { }

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

    [Fact]
    public async Task An_evaluation_on_an_idle_seam_is_submitted_at_once_by_its_own_caller()
    {
        var runner = new BlockingRunner { BlockFirst = false };
        var coalescer = new ScriptBatchCoalescer(runner);
        var outcome = coalescer.RunAsync(Vt("only"));
        outcome.IsCompletedSuccessfully.ShouldBeTrue("submitted synchronously, no timer");
        (await outcome).Value.ShouldBe("only");
        runner.Batches.ToArray().ShouldBe([["only"]]);
    }

    [Fact]
    public async Task Evaluations_submitted_while_a_submission_is_in_progress_go_together_in_arrival_order()
    {
        var runner = new BlockingRunner();
        var coalescer = new ScriptBatchCoalescer(runner);
        var first = Task.Run(() => coalescer.RunAsync(Vt("first")), TestContext.Current.CancellationToken);
        runner.Entered.Wait(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken).ShouldBeTrue();

        var queued = Enumerable.Range(0, 10).Select(i => coalescer.RunAsync(Vt($"k{i}"))).ToArray();
        queued.ShouldAllBe(t => !t.IsCompleted, "queued behind the submission in progress");
        runner.Release.Set();

        (await first).Value.ShouldBe("first");
        var outcomes = await Task.WhenAll(queued);
        outcomes.Select(o => o.Value).ShouldBe(Enumerable.Range(0, 10).Select(i => (object?)$"k{i}"));
        runner.Batches.ToArray().ShouldBe([["first"], [.. Enumerable.Range(0, 10).Select(i => $"k{i}")]]);
    }

    [Fact]
    public async Task A_runner_that_throws_does_not_strand_its_callers()
    {
        var coalescer = new ScriptBatchCoalescer(new ThrowingRunner());
        var outcome = await coalescer.RunAsync(Vt("x"));
        outcome.Status.ShouldBe(ScriptRunStatus.Failed);
        outcome.Reason!.ShouldContain("boom");
        (await coalescer.RunAsync(Vt("y"))).Status.ShouldBe(ScriptRunStatus.Failed, "the flush flag was released");
    }

    private sealed class ThrowingRunner : IScriptRunner
    {
        public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default) => throw new NotSupportedException();

        public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
            => throw new InvalidOperationException("boom");

        public void Reset(ScriptContextKind kind) { }

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