test(scripting): #789 BlockingPipeStream duplex, large frame, peer-gone and self-dispose cases

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 19:13:48 -05:00
parent 3c0e5ea4d7
commit a7fa327e61
1 changed files with 147 additions and 0 deletions
@@ -0,0 +1,147 @@
+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);
+        }
+    }
+}