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

using System.IO.Pipes;
using System.Security.Cryptography;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;

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

/// <summary>
///     lmxopcua#789: <see cref="ScriptWorkerFrameCodec.BlockingStream"/> — on Windows the
///     <see cref="WindowsBlockingPipeStream"/> that keeps worker pipe I/O off the completion port. Exercised on a real
///     pipe pair; on Unix the same tests cover the plain pipe it returns there.
/// </summary>
public sealed class BlockingPipeStreamTests
{
    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    private static async Task<(Stream Server, Stream Client)> PairAsync()
    {
        var name = "otopcua-sw-test-" + Convert.ToHexString(RandomNumberGenerator.GetBytes(6));
        var server = new NamedPipeServerStream(name, PipeDirection.InOut, 1, PipeTransmissionMode.Byte,
            PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
        var client = new NamedPipeClientStream(".", name, PipeDirection.InOut, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
        var accept = server.WaitForConnectionAsync(Ct);
        client.Connect(TimeSpan.FromSeconds(5));
        await accept;
        return (ScriptWorkerFrameCodec.BlockingStream(server), ScriptWorkerFrameCodec.BlockingStream(client));
    }

    private static Envelope Frame(ulong id) => new()
    {
        ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
        Seq = id,
        Heartbeat = new Heartbeat { Inflight = (uint)id },
    };

    [Fact]
    public async Task On_windows_the_pipe_is_wrapped_and_elsewhere_it_is_the_pipe_itself()
    {
        var (server, client) = await PairAsync();
        using (server)
        using (client)
        {
            if (OperatingSystem.IsWindows()) server.ShouldBeOfType<WindowsBlockingPipeStream>();
            else server.ShouldBeAssignableTo<PipeStream>();
        }
    }

    /// <summary>
    ///     Full duplex: each end has a read blocked on a dedicated thread while the other direction keeps writing — a
    ///     handle opened for synchronous I/O would serialise the write behind the blocked read and deadlock here.
    /// </summary>
    [Fact]
    public async Task Reads_blocked_on_both_ends_do_not_hold_up_writes_in_either_direction()
    {
        var (server, client) = await PairAsync();
        using (server)
        using (client)
        {
            const int frames = 2_000;
            Task<int> Pump(Stream from, Stream to)
            {
                var reader = Task.Factory.StartNew(() =>
                {
                    var r = new ScriptWorkerFrameReader(from);
                    var seen = 0;
                    while (seen < frames && r.Read() is { } f)
                    {
                        f.Heartbeat.Inflight.ShouldBe((uint)seen);
                        seen++;
                    }
                    return seen;
                }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                _ = Task.Factory.StartNew(() =>
                {
                    for (var i = 0; i < frames; i++) ScriptWorkerFrameCodec.Write(to, Frame((ulong)i), ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
                }, Ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                return reader;
            }

            var toClient = Pump(client, server);
            var toServer = Pump(server, client);
            (await toClient.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames);
            (await toServer.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldBe(frames);
        }
    }

    [Fact]
    public async Task A_large_frame_crosses_whole()
    {
        var (server, client) = await PairAsync();
        using (server)
        using (client)
        {
            var big = new Envelope
            {
                ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
                Load = new LoadScripts { Scripts = { new ScriptSource { Key = "k", Source = new string('x', 1_000_000) } } },
            };
            var read = Task.Factory.StartNew(() => new ScriptWorkerFrameReader(server).Read(), Ct, TaskCreationOptions.LongRunning,
                TaskScheduler.Default);
            ScriptWorkerFrameCodec.Write(client, big, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
            var got = await read.WaitAsync(TimeSpan.FromSeconds(30), Ct);
            got!.Load.Scripts[0].Source.Length.ShouldBe(1_000_000);
        }
    }

    [Fact]
    public async Task A_peer_that_goes_away_ends_a_blocked_read_and_fails_a_write()
    {
        var (server, client) = await PairAsync();
        using (server)
        {
            var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning,
                TaskScheduler.Default);
            await Task.Delay(100, Ct);
            read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel");
            client.Dispose();
            (await read.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBe(0, "end of stream");
            Should.Throw<IOException>(() =>
            {
                // A pipe can accept a write into its buffer before it notices; the second one must fail.
                for (var i = 0; i < 4; i++) server.Write(new byte[64], 0, 64);
            });
        }
    }

    [Fact]
    public async Task Disposing_an_end_wakes_its_own_blocked_reader()
    {
        var (server, client) = await PairAsync();
        using (client)
        {
            var read = Task.Factory.StartNew(() => server.Read(new byte[16], 0, 16), Ct, TaskCreationOptions.LongRunning,
                TaskScheduler.Default);
            await Task.Delay(100, Ct);
            read.IsCompleted.ShouldBeFalse("positive control: the read is blocked on the kernel");
            server.Dispose();
            // End of stream, or the disposed / I/O error the readers treat as the pipe ending — never a hang.
            int ended;
            try { ended = await read.WaitAsync(TimeSpan.FromSeconds(10), Ct); }
            catch (Exception ex) when (ex is ObjectDisposedException or IOException) { ended = 0; }
            ended.ShouldBe(0);
        }
    }
}