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

using System.Diagnostics;
using System.Security.Cryptography;
using System.Text;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;

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

/// <summary>
///     lmxopcua#789 review: the blocking pipe I/O runs the host's reader on a dedicated thread and blocks producers on a
///     full pipe. A fault in the reader's inline completion work must end that worker, not the server process; and a worker
///     that stops reading while it keeps heartbeating must be killed so blocked producers are released.
/// </summary>
[Collection(ScriptWorkerCollection.Name)]
[Trait("Category", "ScriptWorker")]
public sealed class PipeFaultTests
{
    private const string Healthy = "return 7;";
    private const string Runaway = "while (true) { }";

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

    /// <summary>
    ///     The slot release runs on the pipe reader, inline, before the caller resumes. When it throws, the reader must
    ///     end the worker as a fault: the run in flight beside it answers BadResourceUnavailable ("worker restarted"),
    ///     the shard restarts, and the next evaluation is served — instead of an unhandled exception on a dedicated thread,
    ///     which would terminate the host.
    /// </summary>
    [Fact]
    public async Task A_completion_callback_that_throws_on_the_reader_ends_that_worker_as_faulted_not_the_host()
    {
        await using var runner = Scripts.Runner(Scripts.Options());
        (await runner.RunAsync(Scripts.Vt("boom", Healthy), Ct)).Value.ShouldBe(7, "positive control: the hook is not armed yet");
        var worker = runner.Shards[0].Current.ShouldNotBeNull();

        var spin = runner.RunAsync(Scripts.Vt("spin", Runaway, TimeSpan.FromSeconds(30)), Ct);
        await Scripts.Eventually(() => worker.InFlightCount > 0 ? worker : null, TimeSpan.FromSeconds(30), "the sibling in flight");

        var armed = 1;
        runner.SlotReleasingForTest = key =>
        {
            if (key.EndsWith("boom", StringComparison.Ordinal) && Interlocked.Exchange(ref armed, 0) == 1)
                throw new InvalidOperationException("a completion callback failed");
        };
        await runner.RunAsync(Scripts.Vt("boom", Healthy), Ct);

        var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(15), Ct);
        exit.Reason.ShouldBe(WorkerExitReason.ProtocolFault);
        var sibling = await spin.WaitAsync(TimeSpan.FromSeconds(15), Ct);
        sibling.Status.ShouldBe(ScriptRunStatus.Unavailable);
        sibling.Reason.ShouldBe(ScriptRunOutcome.WorkerRestartedReason);
        sibling.ToVirtualTagEvalResult().StatusCode.ShouldBe(ZB.MOM.WW.OtOpcUa.Commons.OpcUa.OpcUaStatus.BadResourceUnavailable);

        (await runner.RunAsync(Scripts.Vt("after", Healthy), Ct)).Value.ShouldBe(7, "the shard restarted and serves again");
        runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldNotBe(worker.Pid);
    }

    /// <summary>
    ///     A worker whose reader stops while its heartbeats go on: the host's writes fill the pipe and the producer (here
    ///     a deploy loading scripts, the case nothing else would end: no evaluation is outstanding and a compile's timeout
    ///     starts only when the worker starts it) blocks in the write. The write-stall check must kill the worker within
    ///     its bound — as unresponsive, not for silence, which proves it kept heartbeating — and that releases the producer.
    /// </summary>
    [Fact]
    public async Task A_worker_that_stops_reading_but_keeps_heartbeating_is_killed_and_the_blocked_producer_returns()
    {
        var heartbeat = TimeSpan.FromMilliseconds(200);
        var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
            256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: heartbeat,
            HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30), TestStallReaderOnDiag: true);
        var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
        await using var owned = connection;
        var warm = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Healthy,
            SHA256.HashData(Encoding.UTF8.GetBytes(Healthy)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
        warm.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
        connection.SilenceLimit = heartbeat * 5; // armed once the cold start is over (see SupervisionTests)
        var stallLimit = TimeSpan.FromSeconds(2);
        connection.WriteStallLimitOverride = stallLimit;

        _ = connection.DiagnosticsAsync(TimeSpan.FromSeconds(60)); // the worker's reader stops here for good
        var producer = Task.Factory.StartNew(() =>
        {
            // About 5 MB of loads: far more than the pipe's buffer and the worker reader's own.
            for (var i = 0; i < 20; i++)
            {
                var source = $"return {i}; // " + new string('x', 256 * 1024);
                _ = connection.EnsureLoadedAsync($"big{i}", ContextKind.VirtualTag, source,
                    SHA256.HashData(Encoding.UTF8.GetBytes(source)), []);
            }
        }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);

        var sw = Stopwatch.StartNew();
        await Task.Delay(stallLimit / 2, Ct);
        producer.IsCompleted.ShouldBeFalse("positive control: the producer is blocked in a write to the full pipe");
        connection.IsAlive.ShouldBeTrue("positive control: heartbeats keep the silence check quiet");

        var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(20), Ct);
        exit.Reason.ShouldBe(WorkerExitReason.KilledUnresponsive, "killed by the write-stall check, not for silence");
        await producer.WaitAsync(TimeSpan.FromSeconds(10), Ct);
        sw.Elapsed.ShouldBeLessThan(stallLimit + TimeSpan.FromSeconds(8), "the producer was released within the bound");
    }
}