TCP listener: fail a request whose value task cannot be read or awaited

The callback is public and may return a value task over its own source. Reading its state or result, or registering the continuation, can throw (for example for a stale token). Before the pooled continuation, the await was inside the try/finally, so that became a service fault and the request and its pending-request scope were released. Now such an exception escaped to the channel, which only logs it, and the request stayed tracked. Such failures now complete the request like a processing failure. The pooled continuation is claimed atomically: if registration throws after a continuation was registered, only one of them completes the request, and the object is not returned to the pool. The slow path takes an explicit flag for whether the processing was already read, so a null response is not read twice. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 21:02:51 +08:00 · 2026-10-15 08:25:34 +02:00
parent 39bfb8a931
commit ee81c83f2b
2 changed files with 172 additions and 18 deletions
@@ -1742,15 +1742,33 @@ namespace Opc.Ua.Bindings
            }
            catch (Exception e)
            {
-                processing = new ValueTask<IServiceResponse>(Task.FromException<IServiceResponse>(e));
+                processing = Faulted(e);
            }

-            if (processing.IsCompleted)
+            // The callback may return a value task over its own source, which can throw
+            // when its state is read or a continuation is registered: the request then
+            // fails like one whose processing threw.
+            bool completed;
+            try
+            {
+                completed = processing.IsCompleted;
+            }
+            catch (Exception e)
+            {
+                processing = Faulted(e);
+                completed = true;
+            }
+            if (completed)
            {
                OnRequestProcessed(channel, requestId, request, usage, processing);
                return;
            }
-            s_pendingRequests.Get().Await(this, channel, requestId, request, usage, processing);
+            if (!s_pendingRequests.Get().TryAwait(
+                    this, channel, requestId, request, usage, processing, out Exception? error) &&
+                error != null)
+            {
+                CompleteRequestAsync(channel, requestId, request, usage, Faulted(error), null, default, false);
+            }
        }

        /// <summary>
@@ -1764,17 +1782,34 @@ namespace Opc.Ua.Bindings
            TcpListenerChannel.PendingRequestScope usage,
            ValueTask<IServiceResponse> processing)
        {
-            if (!processing.IsCompletedSuccessfully)
+            IServiceResponse? response = null;
+            bool succeeded = false;
+            Exception? readError = null;
+            try
            {
-                CompleteRequestAsync(channel, requestId, request, usage, processing, null, default);
+                if (processing.IsCompletedSuccessfully)
+                {
+                    response = processing.Result;
+                    succeeded = true;
+                }
+            }
+            catch (Exception e)
+            {
+                // a value task source of the callback that cannot report its result
+                readError = e;
+            }
+            if (!succeeded)
+            {
+                CompleteRequestAsync(
+                    channel, requestId, request, usage,
+                    readError == null ? processing : Faulted(readError), null, default, false);
                return;
            }

-            IServiceResponse response = processing.Result;
            ValueTask<bool> sending;
            try
            {
-                sending = ((TcpServerChannel)channel).SendResponseAsync(requestId, response);
+                sending = ((TcpServerChannel)channel).SendResponseAsync(requestId, response!);
            }
            catch (Exception e)
            {
@@ -1783,7 +1818,7 @@ namespace Opc.Ua.Bindings

            if (!sending.IsCompletedSuccessfully)
            {
-                CompleteRequestAsync(channel, requestId, request, usage, default, response, sending);
+                CompleteRequestAsync(channel, requestId, request, usage, default, response, sending, true);
                return;
            }

@@ -1808,9 +1843,10 @@ namespace Opc.Ua.Bindings
        /// <param name="request">The request.</param>
        /// <param name="usage">The pending-request scope, disposed when the request completes.</param>
        /// <param name="processing">The completed processing of the request, unless
-        /// <paramref name="response"/> is set.</param>
+        /// <paramref name="sendStarted"/>.</param>
        /// <param name="response">The response whose send is <paramref name="sending"/>.</param>
        /// <param name="sending">The send of <paramref name="response"/>.</param>
+        /// <param name="sendStarted">Whether the processing was read and its send started.</param>
        private async void CompleteRequestAsync(
            TcpListenerChannel channel,
            uint requestId,
@@ -1818,13 +1854,14 @@ namespace Opc.Ua.Bindings
            TcpListenerChannel.PendingRequestScope usage,
            ValueTask<IServiceResponse> processing,
            IServiceResponse? response,
-            ValueTask<bool> sending)
+            ValueTask<bool> sending,
+            bool sendStarted)
        {
            using TcpListenerChannel.PendingRequestScope scope = usage;
            bool responseRetained = false;
            try
            {
-                if (response == null)
+                if (!sendStarted)
                {
                    response = await processing.ConfigureAwait(false);
                    sending = ((TcpServerChannel)channel).SendResponseAsync(requestId, response);
@@ -1850,7 +1887,7 @@ namespace Opc.Ua.Bindings
                        if (serverChannel != channel)
                        {
                            responseRetained = await serverChannel
-                                .SendResponseAsync(requestId, response).ConfigureAwait(false);
+                                .SendResponseAsync(requestId, response!).ConfigureAwait(false);
                            return;
                        }
                    }
@@ -1913,6 +1950,14 @@ namespace Opc.Ua.Bindings
            }
        }

+        /// <summary>
+        /// A failed processing, for a request whose processing could not be started or read.
+        /// </summary>
+        private static ValueTask<IServiceResponse> Faulted(Exception error)
+        {
+            return new ValueTask<IServiceResponse>(Task.FromException<IServiceResponse>(error));
+        }
+
        /// <summary>
        /// Awaits the processing of a request in place of an async state machine per request.
        /// Pooled: it returns itself to the pool before the response is sent.
@@ -1926,33 +1971,63 @@ namespace Opc.Ua.Bindings

            /// <summary>
            /// Continues with <see cref="OnRequestProcessed"/> once the processing completes.
+            /// Returns <c>false</c> with the error when the continuation cannot be registered;
+            /// the caller then completes the request, unless the error is <c>null</c> because
+            /// a continuation registered before the failure already did.
            /// </summary>
-            public void Await(
+            public bool TryAwait(
                TcpTransportListener listener,
                TcpListenerChannel channel,
                uint requestId,
                IServiceRequest request,
                TcpListenerChannel.PendingRequestScope usage,
-                ValueTask<IServiceResponse> processing)
+                ValueTask<IServiceResponse> processing,
+                out Exception? error)
            {
-                m_listener = listener;
                m_channel = channel;
                m_requestId = requestId;
                m_request = request;
                m_usage = usage;
                m_processing = processing;
-                processing.ConfigureAwait(false).GetAwaiter().OnCompleted(m_onProcessed);
+                // published last: the continuation claims the request by taking it
+                Volatile.Write(ref m_listener, listener);
+                try
+                {
+                    processing.ConfigureAwait(false).GetAwaiter().OnCompleted(m_onProcessed);
+                    error = null;
+                    return true;
+                }
+                catch (Exception e)
+                {
+                    // Not returned to the pool: a continuation registered before the
+                    // failure may still run, and must find the request claimed.
+                    if (Interlocked.Exchange(ref m_listener, null) == null)
+                    {
+                        error = null;
+                        return false;
+                    }
+                    m_channel = null;
+                    m_request = null;
+                    m_usage = default;
+                    m_processing = default;
+                    error = e;
+                    return false;
+                }
            }

            private void OnProcessed()
            {
-                TcpTransportListener listener = m_listener!;
+                TcpTransportListener? listener = Interlocked.Exchange(ref m_listener, null);
+                if (listener == null)
+                {
+                    // the registration failed and the request was completed without it
+                    return;
+                }
                TcpListenerChannel channel = m_channel!;
                uint requestId = m_requestId;
                IServiceRequest request = m_request!;
                TcpListenerChannel.PendingRequestScope usage = m_usage;
                ValueTask<IServiceResponse> processing = m_processing;
-                m_listener = null;
                m_channel = null;
                m_request = null;
                m_usage = default;
@@ -35,6 +35,7 @@ using System.Net;
 using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Threading.Tasks.Sources;
 using Microsoft.Extensions.Logging;
 using Moq;
 using NUnit.Framework;
@@ -193,6 +194,35 @@ namespace Opc.Ua.Core.Tests.Stack.Transport
            Assert.That(next.ResponseHeader.RequestHandle, Is.EqualTo(2));
        }

+        /// <summary>
+        /// A callback's value task source that throws when the listener reads its state,
+        /// registers the continuation or reads the result fails that request like a processing
+        /// failure: logged, answered with a service fault, and the channel keeps working.
+        /// </summary>
+        [TestCase(ThrowingSourceMode.GetStatus)]
+        [TestCase(ThrowingSourceMode.OnCompleted)]
+        [TestCase(ThrowingSourceMode.GetResult)]
+        [CancelAfter(30000)]
+        public async Task ThrowingValueTaskSourceIsAnsweredWithServiceFaultAsync(ThrowingSourceMode mode)
+        {
+            var callback = new ScriptedCallback(request => request.RequestHeader.RequestHandle == 1
+                ? new ValueTask<IServiceResponse>(new ThrowingSource(mode), 0)
+                : new ValueTask<IServiceResponse>(Respond(request)));
+            await using Loopback loopback = await Loopback.StartAsync(callback).ConfigureAwait(false);
+            using UaSCUaBinaryTransportChannel client = await loopback.ConnectAsync().ConfigureAwait(false);
+
+            IServiceResponse fault = await client.SendRequestAsync(CreateReadRequest(1), CancellationToken.None)
+                .ConfigureAwait(false);
+
+            Assert.That(fault, Is.InstanceOf<ServiceFault>());
+            Assert.That(fault.ResponseHeader.ServiceResult, Is.EqualTo(StatusCodes.BadNotSupported));
+            Assert.That(loopback.Logs.Count(l => l.Level == LogLevel.Error &&
+                l.EventId == CoreEventIds.TcpTransportListener + 22), Is.EqualTo(1));
+            IServiceResponse next = await client.SendRequestAsync(CreateReadRequest(2), CancellationToken.None)
+                .ConfigureAwait(false);
+            Assert.That(next.ResponseHeader.RequestHandle, Is.EqualTo(2));
+        }
+
        /// <summary>
        /// A cancelled request is answered with the fault the endpoint creates for it.
        /// </summary>
@@ -271,6 +301,55 @@ namespace Opc.Ua.Core.Tests.Stack.Transport
            LaterFault
        }

+        public enum ThrowingSourceMode
+        {
+            GetStatus,
+            OnCompleted,
+            GetResult
+        }
+
+        /// <summary>
+        /// A value task source that fails at one step of being awaited.
+        /// </summary>
+        private sealed class ThrowingSource : IValueTaskSource<IServiceResponse>
+        {
+            public ThrowingSource(ThrowingSourceMode mode)
+            {
+                m_mode = mode;
+            }
+
+            public ValueTaskSourceStatus GetStatus(short token)
+            {
+                return m_mode switch
+                {
+                    ThrowingSourceMode.GetStatus => throw Failure(),
+                    ThrowingSourceMode.OnCompleted => ValueTaskSourceStatus.Pending,
+                    _ => ValueTaskSourceStatus.Succeeded
+                };
+            }
+
+            public void OnCompleted(
+                Action<object?> continuation,
+                object? state,
+                short token,
+                ValueTaskSourceOnCompletedFlags flags)
+            {
+                throw Failure();
+            }
+
+            public IServiceResponse GetResult(short token)
+            {
+                throw Failure();
+            }
+
+            private static ServiceResultException Failure()
+            {
+                return new ServiceResultException(StatusCodes.BadNotSupported, "Injected source failure.");
+            }
+
+            private readonly ThrowingSourceMode m_mode;
+        }
+
        private void OnUnobservedTaskException(object? sender, UnobservedTaskExceptionEventArgs e)
        {
            m_unobserved.Enqueue(e.Exception);