From fe6ba606c382aa9fabbeaa13c818a6992bc6b801 Mon Sep 17 00:00:00 2001 From: romanett Date: Fri, 2 Oct 2026 21:23:28 +0200 Subject: [PATCH] Fail sends racing PipeByteTransport.Close with BadConnectionClosed (#4643) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit # Description Fixes an intermittent failure of `PipeByteTransportTests.CloseFailsBlockedAndQueuedSendsWithBadConnectionClosedAsync`, seen on windows-net10.0 in run [37025190491](https://github.com/OPCFoundation/UA-.NETStandard/actions/runs/37025190491) (job 110899124242) on #4629, a PR that does not touch this code: ``` Expected: But was: System.InvalidOperationException: Writing is not allowed after writer was completed. at System.IO.Pipelines.Pipe.DefaultPipeWriter.GetMemory(Int32 sizeHint) at Opc.Ua.Bindings.PipeByteTransport.SendChunkAsync(...) PipeByteTransport.cs:line 101 ``` **Cause:** `PipeByteTransport.Close()` completes the output `PipeWriter` without taking the send lock. A send can pass the closed check and get the lock just before `Close()` runs. In the test this is most likely the queued send: the blocked send releases the lock as it is cancelled, and the queued one gets it before its own cancellation takes effect. That send then calls `GetMemory` on the completed writer and leaks the pipe's `InvalidOperationException`. **Fix:** `Close()` sets the closed flag before it completes the writer. So when a write fails while the transport is closed, both `SendChunkAsync` overloads now turn the failure into `ServiceResultException(BadConnectionClosed)` and keep the original exception as the inner exception. `ServiceResultException` and `OperationCanceledException` are excluded, so cancellation by the caller's own token still surfaces as before. This translates the exception instead of taking the lock in `Close()`: a check before writing would still race, and taking the lock would stop `Close()` from waking a send blocked on backpressure. **Tests:** two new tests, one per `SendChunkAsync` overload, wrap the output writer so `GetMemory` calls `transport.Close()` first. This hits the race every time. Both failed before the fix with the exact CI exception and pass after it. All 28 `PipeByteTransportTests` pass on net10.0, and the original flaky test passed 30/30 in a loop. ## Related Issues - Flaky test seen on #4629 (CI run 37025190491) ## Checklist - [ ] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [x] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [x] I have added all necessary documentation. - [x] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../Tcp/PipeByteTransport.cs | 29 ++++++ .../PipeByteTransportTests.cs | 96 +++++++++++++++++++ 2 files changed, 125 insertions(+) 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()