lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverBatchPumpTests.cs

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

namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests;

/// <summary>
///     lmxopcua#780 batch submit (design §4.5) in the Calculation pump: with an out-of-process runner the pump hands
///     everything it dequeued to ONE <see cref="IScriptRunner.RunBatch"/> and publishes in dequeue order. A tag queued
///     twice ends the batch, so every queued evaluation still runs, in order, against the snapshot that triggered it —
///     the published sequence is exactly the one-at-a-time pump's.
/// </summary>
public sealed class CalculationDriverBatchPumpTests
{
    private const uint StatusGood = 0u;

    /// <summary>
    ///     Answers <c>src/a × n</c> for tag <c>calc/tn</c>. The first batch is held until <see cref="Release"/>, so the
    ///     test can queue more triggers behind it.
    /// </summary>
    private sealed class GatedRunner : IScriptRunner
    {
        private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private int _batches;

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

        public int SingleRuns;

        public void Release() => _gate.SetResult();

        public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
        {
            Interlocked.Increment(ref SingleRuns);
            return Task.FromResult(Answer(invocation));
        }

        public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
        {
            Batches.Enqueue(invocations.Select(i => i.Key).ToArray());
            var held = Interlocked.Increment(ref _batches) == 1;
            return invocations.Select(i => held ? Later(i) : Task.FromResult(Answer(i))).ToArray();
        }

        private async Task<ScriptRunOutcome> Later(ScriptInvocation invocation)
        {
            await _gate.Task.ConfigureAwait(false);
            return Answer(invocation);
        }

        private static ScriptRunOutcome Answer(ScriptInvocation invocation)
            => ScriptRunOutcome.Ok((double)invocation.Inputs["src/a"].Value! * int.Parse(invocation.Key["calc/t".Length..]));

        public void Reset(ScriptContextKind kind) { }

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

    private static RawTagEntry CalcTag(string rawPath, int factor)
    {
        var source = $"return (double)ctx.GetTag(\"src/a\").Value * {factor};";
        var json = $"{{\"scriptId\":\"s{factor}\",\"changeTriggered\":true,\"scriptSource\":{System.Text.Json.JsonSerializer.Serialize(source)}}}";
        return new RawTagEntry(rawPath, json, WriteIdempotent: false);
    }

    private static async Task Settle(CalculationDriver driver)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30);
        while (DateTime.UtcNow < deadline && !driver.IsEvaluationIdle) await Task.Delay(10);
        driver.IsEvaluationIdle.ShouldBeTrue();
    }

    [Fact]
    public async Task The_pump_submits_what_it_dequeued_as_one_batch_and_publishes_in_dequeue_order()
    {
        var runner = new GatedRunner();
        var driver = new CalculationDriver(
            new CalculationDriverOptions { RawTags = [CalcTag("calc/t1", 1), CalcTag("calc/t2", 2), CalcTag("calc/t3", 3)], ScriptRunner = runner },
            "calc-drv", new ScriptRootLogger(new LoggerConfiguration().CreateLogger()));
        var published = new ConcurrentQueue<(string Tag, object? Value, uint Status)>();
        driver.OnDataChange += (_, e) => published.Enqueue((e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode));
        var sizes = new ConcurrentQueue<int>();
        driver.BatchSubmittedForTest = sizes.Enqueue;
        await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);

        driver.OnDependencyValue("src/a", 1.0, StatusGood, DateTime.UtcNow);
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
        while (DateTime.UtcNow < deadline && runner.Batches.IsEmpty) await Task.Delay(5, TestContext.Current.CancellationToken);
        // Two more triggers queue behind the held batch: six evaluations, each tag twice.
        driver.OnDependencyValue("src/a", 2.0, StatusGood, DateTime.UtcNow);
        driver.OnDependencyValue("src/a", 3.0, StatusGood, DateTime.UtcNow);
        runner.Release();
        await Settle(driver);

        runner.SingleRuns.ShouldBe(0);
        runner.Batches.ToArray().ShouldBe(
            [["calc/t1", "calc/t2", "calc/t3"], ["calc/t1", "calc/t2", "calc/t3"], ["calc/t1", "calc/t2", "calc/t3"]],
            "a tag queued twice ends the batch; every queued evaluation still runs");
        sizes.ToArray().ShouldBe([3, 3, 3]);
        published.ToArray().ShouldBe(
        [
            ("calc/t1", 1.0, StatusGood), ("calc/t2", 2.0, StatusGood), ("calc/t3", 3.0, StatusGood),
            ("calc/t1", 2.0, StatusGood), ("calc/t2", 4.0, StatusGood), ("calc/t3", 6.0, StatusGood),
            ("calc/t1", 3.0, StatusGood), ("calc/t2", 6.0, StatusGood), ("calc/t3", 9.0, StatusGood),
        ], "dequeue order, each against the snapshot that triggered it");
    }
}