test(ab-transport): blocked-write timeout fault path via a stream seam (#565, #593)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs

@@ -39,7 +39,7 @@ public sealed class EncapsulationClient : IAsyncDisposable
     private static readonly TimeSpan UnregisterBudget = TimeSpan.FromMilliseconds(500);
 
     private readonly TcpClient _tcp;
-    private readonly NetworkStream _stream;
+    private readonly Stream _stream;
 
     // Never disposed: an abandoned exchange may release it after DisposeAsync returns, and
     // SemaphoreSlim.Release on a disposed instance throws. It holds no unmanaged resource unless
@@ -56,12 +56,17 @@ public sealed class EncapsulationClient : IAsyncDisposable
     /// <summary>False once the connection has faulted or been disposed.</summary>
     public bool IsConnected => !_faulted && Volatile.Read(ref _disposed) == 0 && _tcp.Connected;
 
-    private EncapsulationClient(TcpClient tcp)
+    private EncapsulationClient(TcpClient tcp, Stream? stream = null)
     {
         _tcp = tcp;
-        _stream = tcp.GetStream();
+        _stream = stream ?? tcp.GetStream();
     }
 
+    /// <summary>Test seam: wraps an already-connected <paramref name="tcp"/> but routes all frame I/O through
+    /// <paramref name="stream"/> (e.g. one whose write never completes), skipping RegisterSession. The TCP
+    /// client is still what <see cref="Fault"/> closes and <see cref="IsConnected"/> consults.</summary>
+    internal static EncapsulationClient ForStream(TcpClient tcp, Stream stream) => new(tcp, stream);
+
     /// <summary>Open the TCP connection and register an encapsulation session. <paramref name="timeout"/>
     /// bounds the TCP connect and the RegisterSession exchange separately.</summary>
     public static async Task<EncapsulationClient> ConnectAsync(string host, int port, TimeSpan timeout, CancellationToken ct)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs

@@ -295,6 +295,74 @@ public sealed class EncapsulationClientTests
         client.IsConnected.ShouldBeFalse();
     }
 
+    /// <summary>A stream whose write never completes until released, standing in for a peer that stopped reading.</summary>
+    private sealed class BlockedWriteStream : Stream
+    {
+        private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource _writeStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public Task WriteStarted => _writeStarted.Task;
+        public void FailPendingWrite() => _release.TrySetException(new IOException("socket closed"));
+
+        public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            _writeStarted.TrySetResult();
+            await _release.Task.ConfigureAwait(false);   // ignores cancellation, like a stuck socket send
+        }
+
+        public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) =>
+            throw new InvalidOperationException("The exchange must never reach the read side while its write is blocked.");
+        public override bool CanRead => true;
+        public override bool CanSeek => false;
+        public override bool CanWrite => true;
+        public override long Length => throw new NotSupportedException();
+        public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
+        public override void Flush() { }
+        public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+        public override void SetLength(long value) => throw new NotSupportedException();
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+    }
+
+    [Fact]
+    public async Task A_blocked_write_times_out_faults_the_session_and_leaves_no_unobserved_exception()
+    {
+        // Gitea #565 / #593: the write ignores cancellation (a half-written frame would desync the stream) but is
+        // bounded. A stream whose write never completes is the deterministic form of "peer stopped reading".
+        var unobserved = await UnobservedTaskProbe.CollectAsync(AbandonABlockedWrite, Guard);
+        unobserved.ShouldNotContain(x => x.Message.Contains("socket closed"));
+    }
+
+    [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)]
+    private static async Task AbandonABlockedWrite()
+    {
+        using var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        using var tcp = new TcpClient();
+        var accept = listener.AcceptTcpClientAsync(TestCt).AsTask();
+        await tcp.ConnectAsync(IPAddress.Loopback, port, TestCt);
+        using var peer = await accept;
+
+        var stream = new BlockedWriteStream();
+        await using var client = EncapsulationClient.ForStream(tcp, stream);
+        client.IsConnected.ShouldBeTrue();
+
+        var call = client.SendRRDataAsync(new byte[] { 0x01 }, TimeSpan.FromMilliseconds(250), TestCt);
+        await stream.WriteStarted.WaitAsync(Guard, TestCt);   // positive control: the write really was attempted
+        var ex = await Should.ThrowAsync<AbTransportException>(call.WaitAsync(Guard, TestCt));
+        ex.Status.ShouldBe(AbStatus.Timeout);
+        client.IsConnected.ShouldBeFalse();
+
+        // Faulted: the next call is refused up front rather than queued behind the stuck write.
+        var next = await Should.ThrowAsync<AbTransportException>(
+            client.SendRRDataAsync(new byte[] { 0x01 }, Generous, TestCt));
+        next.Status.ShouldBe(AbStatus.BadConnection);
+
+        // Closing the socket is what fails the pending write in production; the abandoned write task then faults
+        // and must have been observed.
+        stream.FailPendingWrite();
+    }
+
     [Fact]
     public async Task Connected_sequence_is_assigned_inside_the_gate()
     {