diff --git a/src/Opc.Ua.Bindings.Https/Tcp/PipeByteTransport.cs b/src/Opc.Ua.Bindings.Https/Tcp/PipeByteTransport.cs index ff0c1635fa..64d4cebdab 100644 --- a/src/Opc.Ua.Bindings.Https/Tcp/PipeByteTransport.cs +++ b/src/Opc.Ua.Bindings.Https/Tcp/PipeByteTransport.cs @@ -115,6 +115,13 @@ namespace Opc.Ua.Bindings StatusCodes.BadConnectionClosed, "Transport closed while writing chunk."); } + catch (Exception ex) when (IsWriterClosedByClose(ex)) + { + throw ServiceResultException.Create( + StatusCodes.BadConnectionClosed, + ex, + "Transport closed while writing chunk."); + } finally { m_sendLock.Release(); @@ -155,6 +162,13 @@ namespace Opc.Ua.Bindings StatusCodes.BadConnectionClosed, "Transport closed while writing buffer collection."); } + catch (Exception ex) when (IsWriterClosedByClose(ex)) + { + throw ServiceResultException.Create( + StatusCodes.BadConnectionClosed, + ex, + "Transport closed while writing buffer collection."); + } finally { m_sendLock.Release(); @@ -356,6 +370,21 @@ namespace Opc.Ua.Bindings } } + /// + /// A sender that holds the send lock can race , + /// which completes the output writer without taking the lock. The + /// writer then rejects the write (InvalidOperationException from a + /// completed pipe) and that must surface as + /// . Close sets the + /// closed flag before completing the writer, so the flag is set + /// whenever the writer fails for that reason. + /// + private bool IsWriterClosedByClose(Exception ex) + { + return ex is not ServiceResultException and not OperationCanceledException && + Volatile.Read(ref m_closed) != 0; + } + private void ThrowIfClosed() { if (Volatile.Read(ref m_closed) != 0) diff --git a/tests/Opc.Ua.Bindings.Https.WebApi.Tests/PipeByteTransportTests.cs b/tests/Opc.Ua.Bindings.Https.WebApi.Tests/PipeByteTransportTests.cs index 451fc6a70d..5c17eeb0d6 100644 --- a/tests/Opc.Ua.Bindings.Https.WebApi.Tests/PipeByteTransportTests.cs +++ b/tests/Opc.Ua.Bindings.Https.WebApi.Tests/PipeByteTransportTests.cs @@ -536,6 +536,54 @@ namespace Opc.Ua.Bindings.Https.WebApi.Tests } } + /// + /// A send that already holds the send lock when the transport closes + /// reaches on a completed writer. + /// It must fail with + /// instead of leaking the writer's . + /// + [Test] + public void SendChunkAsyncRacingCloseThrowsBadConnectionClosed() + { + using var ctx = new TestConnectionContext(); + var transport = new PipeByteTransport(ctx, m_bufferManager, kBufferSize, m_telemetry); + ctx.Transport = new DuplexPipeAdapter( + ctx.Transport.Input, + new CloseOnGetMemoryPipeWriter(ctx.Transport.Output, transport.Close)); + + ServiceResultException ex = Assert.ThrowsAsync( + async () => await transport.SendChunkAsync( + new ReadOnlyMemory(new byte[8]), + CancellationToken.None).ConfigureAwait(false))!; + Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadConnectionClosed)); + Assert.That(ex.InnerException, Is.TypeOf()); + } + + /// + /// The send racing close must also fail + /// with . + /// + [Test] + public void SendChunkAsyncBufferCollectionRacingCloseThrowsBadConnectionClosed() + { + using var ctx = new TestConnectionContext(); + var transport = new PipeByteTransport(ctx, m_bufferManager, kBufferSize, m_telemetry); + ctx.Transport = new DuplexPipeAdapter( + ctx.Transport.Input, + new CloseOnGetMemoryPipeWriter(ctx.Transport.Output, transport.Close)); + var buffers = new BufferCollection + { + new ArraySegment(new byte[8]) + }; + + ServiceResultException ex = Assert.ThrowsAsync( + async () => await transport.SendChunkAsync( + buffers, + CancellationToken.None).ConfigureAwait(false))!; + Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadConnectionClosed)); + Assert.That(ex.InnerException, Is.TypeOf()); + } + /// /// Helper: build a valid UASC chunk with a Hello message type and /// the given total (header + body). @@ -647,6 +695,54 @@ namespace Opc.Ua.Bindings.Https.WebApi.Tests public PipeWriter Output { get; } } + /// + /// that runs a callback before the first + /// , reproducing a close that lands after the + /// sender passed the closed check and acquired the send lock. + /// + private sealed class CloseOnGetMemoryPipeWriter : PipeWriter + { + public CloseOnGetMemoryPipeWriter(PipeWriter inner, Action onGetMemory) + { + m_inner = inner; + m_onGetMemory = onGetMemory; + } + + public override Memory GetMemory(int sizeHint = 0) + { + Interlocked.Exchange(ref m_onGetMemory, null)?.Invoke(); + return m_inner.GetMemory(sizeHint); + } + + public override Span GetSpan(int sizeHint = 0) + { + return GetMemory(sizeHint).Span; + } + + public override void Advance(int bytes) + { + m_inner.Advance(bytes); + } + + public override void CancelPendingFlush() + { + m_inner.CancelPendingFlush(); + } + + public override void Complete(Exception? exception = null) + { + m_inner.Complete(exception); + } + + public override ValueTask FlushAsync(CancellationToken cancellationToken = default) + { + return m_inner.FlushAsync(cancellationToken); + } + + private readonly PipeWriter m_inner; + private Action? m_onGetMemory; + } + private sealed class TelemetryStub : TelemetryContextBase { public TelemetryStub()