diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs index 571b6ce2..2412d8d0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs +++ b/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 /// False once the connection has faulted or been disposed. 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(); } + /// Test seam: wraps an already-connected but routes all frame I/O through + /// (e.g. one whose write never completes), skipping RegisterSession. The TCP + /// client is still what closes and consults. + internal static EncapsulationClient ForStream(TcpClient tcp, Stream stream) => new(tcp, stream); + /// Open the TCP connection and register an encapsulation session. /// bounds the TCP connect and the RegisterSession exchange separately. public static async Task ConnectAsync(string host, int port, TimeSpan timeout, CancellationToken ct) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs index e70fb159..dc806427 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs @@ -295,6 +295,74 @@ public sealed class EncapsulationClientTests client.IsConnected.ShouldBeFalse(); } + /// A stream whose write never completes until released, standing in for a peer that stopped reading. + 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 buffer, CancellationToken cancellationToken = default) + { + _writeStarted.TrySetResult(); + await _release.Task.ConfigureAwait(false); // ignores cancellation, like a stuck socket send + } + + public override ValueTask ReadAsync(Memory 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(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( + 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() {