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; /// /// 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. /// [Trait("Category", "Unit")] public sealed class ScriptBatchCoalescerTests { private static readonly IReadOnlyDictionary NoInputs = new Dictionary(); private static ScriptInvocation Vt(string key) => new(key, ScriptContextKind.VirtualTag, "return 1;", NoInputs, TimeSpan.FromSeconds(1)); /// Answers each key with itself; the first blocks until released. private sealed class BlockingRunner : IScriptRunner { private int _calls; public ManualResetEventSlim Entered { get; } = new(); public ManualResetEventSlim Release { get; } = new(); public ConcurrentQueue Batches { get; } = new(); public bool BlockFirst { get; init; } = true; public Task RunAsync(ScriptInvocation invocation, CancellationToken ct = default) => throw new InvalidOperationException("the coalescer must submit through RunBatch"); public IReadOnlyList> RunBatch(IReadOnlyList 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 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 RunAsync(ScriptInvocation invocation, CancellationToken ct = default) => throw new NotSupportedException(); public IReadOnlyList> RunBatch(IReadOnlyList invocations, CancellationToken ct = default) => throw new InvalidOperationException("boom"); public void Reset(ScriptContextKind kind) { } public IReadOnlyCollection QuarantinedKeys(ScriptContextKind kind) => []; } }