mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-17 05:11:13 +08:00
Fail sends racing PipeByteTransport.Close with BadConnectionClosed (#4643)
# 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: <ServiceResultException> 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A sender that holds the send lock can race <see cref="Close"/>,
|
||||
/// 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
|
||||
/// <see cref="StatusCodes.BadConnectionClosed"/>. Close sets the
|
||||
/// closed flag before completing the writer, so the flag is set
|
||||
/// whenever the writer fails for that reason.
|
||||
/// </summary>
|
||||
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)
|
||||
|
||||
@@ -536,6 +536,54 @@ namespace Opc.Ua.Bindings.Https.WebApi.Tests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A send that already holds the send lock when the transport closes
|
||||
/// reaches <see cref="PipeWriter.GetMemory"/> on a completed writer.
|
||||
/// It must fail with <see cref="StatusCodes.BadConnectionClosed"/>
|
||||
/// instead of leaking the writer's <see cref="InvalidOperationException"/>.
|
||||
/// </summary>
|
||||
[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<ServiceResultException>(
|
||||
async () => await transport.SendChunkAsync(
|
||||
new ReadOnlyMemory<byte>(new byte[8]),
|
||||
CancellationToken.None).ConfigureAwait(false))!;
|
||||
Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadConnectionClosed));
|
||||
Assert.That(ex.InnerException, Is.TypeOf<InvalidOperationException>());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The <see cref="BufferCollection"/> send racing close must also fail
|
||||
/// with <see cref="StatusCodes.BadConnectionClosed"/>.
|
||||
/// </summary>
|
||||
[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<byte>(new byte[8])
|
||||
};
|
||||
|
||||
ServiceResultException ex = Assert.ThrowsAsync<ServiceResultException>(
|
||||
async () => await transport.SendChunkAsync(
|
||||
buffers,
|
||||
CancellationToken.None).ConfigureAwait(false))!;
|
||||
Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadConnectionClosed));
|
||||
Assert.That(ex.InnerException, Is.TypeOf<InvalidOperationException>());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Helper: build a valid UASC chunk with a Hello message type and
|
||||
/// the given total <paramref name="size"/> (header + body).
|
||||
@@ -647,6 +695,54 @@ namespace Opc.Ua.Bindings.Https.WebApi.Tests
|
||||
public PipeWriter Output { get; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="PipeWriter"/> that runs a callback before the first
|
||||
/// <see cref="GetMemory"/>, reproducing a close that lands after the
|
||||
/// sender passed the closed check and acquired the send lock.
|
||||
/// </summary>
|
||||
private sealed class CloseOnGetMemoryPipeWriter : PipeWriter
|
||||
{
|
||||
public CloseOnGetMemoryPipeWriter(PipeWriter inner, Action onGetMemory)
|
||||
{
|
||||
m_inner = inner;
|
||||
m_onGetMemory = onGetMemory;
|
||||
}
|
||||
|
||||
public override Memory<byte> GetMemory(int sizeHint = 0)
|
||||
{
|
||||
Interlocked.Exchange(ref m_onGetMemory, null)?.Invoke();
|
||||
return m_inner.GetMemory(sizeHint);
|
||||
}
|
||||
|
||||
public override Span<byte> 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<FlushResult> 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()
|
||||
|
||||
Reference in New Issue
Block a user