Complete each pooled service response at most once (#4729) (#4736)

## Proposed changes

Fixes the cross-talk part of #4729 on master.

`ServiceResponsePooledValueTaskSource` comes from one pool shared by the
whole process, and `ManualResetValueTaskSourceCore.SetResult` never
checks the version token. A second completion of an
`EndpointIncomingRequest` therefore either threw
`InvalidOperationException` or, once the source had been read and
recycled, answered whichever request rented it next. That request could
belong to another server instance in the same process.

master's request queue (Channel + `FairRequestQueue`) already completes
each request exactly once. But `ServerBase.ScheduleIncomingRequest` and
`ProcessRequestAsync` are virtual, so a single override that completes
and then throws (the `ProcessAsync` catch completes again) was enough to
corrupt unrelated requests.

- Each rental of the pooled source now gets a lease number, which is
never reused for a later rental of the same instance.
- `EndpointIncomingRequest` captures the lease when it is constructed
and completes through `TrySetResult(lease, response)`. That is a single
CAS on the pending lease, so only the first completion of the current
rental is accepted. Late or stale completions are discarded and can
never reach the next owner.
- `SetResult`/`SetException`, which had no callers outside this path,
are replaced by `TrySetResult`.

The master378 backport #4738 also fixes the double completions in the
old `RequestQueue` itself (snapshot in `Dispose`, and enqueue followed
by a throw).

## Tests
- `ServiceResponsePooledValueTaskSourceTests`: a second completion of
the same rental is ignored, and a late completion after the source is
recycled does not complete the next rental. Both use a private pool, so
the result is deterministic.
-
`EndpointIncomingRequestTests.ProcessAsync_ScheduleCompletesThenThrows_KeepsFirstCompletionAsync`:
a scheduler that completes the request and then throws. Without the fix
it fails with `InvalidOperationException`.
- `Opc.Ua.Core.Tests` Stack.Server + Stack.Tcp (net10.0): 1871 passed, 0
failed.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/Opc.Ua.Core/Stack/Server/EndpointBase.EndpointIncomingRequest.cs

@@ -56,6 +56,7 @@ namespace Opc.Ua
                     ? new RequestParkSink()
                     : null;
                 m_vts = ServiceResponsePooledValueTaskSource.Create();
+                m_lease = m_vts.Lease;
                 m_transportCancellationToken = default;
             }
 
@@ -81,7 +82,7 @@ namespace Opc.Ua
                 }
                 catch (Exception e)
                 {
-                    m_vts.SetResult(m_endpoint.CreateFault(Request, e));
+                    m_vts.TrySetResult(m_lease, m_endpoint.CreateFault(Request, e));
                 }
 
                 return m_vts.Task;
@@ -144,7 +145,7 @@ namespace Opc.Ua
                                 .ConfigureAwait(false);
                         }
 
-                        m_vts.SetResult(response);
+                        m_vts.TrySetResult(m_lease, response);
                     }
                 }
                 catch (Exception e)
@@ -154,7 +155,7 @@ namespace Opc.Ua
                         e = new ServiceResultException(
                             requestLifetime.StatusCode == StatusCodes.Good ? StatusCodes.BadTimeout : requestLifetime.StatusCode);
                     }
-                    m_vts.SetResult(m_endpoint.CreateFault(Request, e));
+                    m_vts.TrySetResult(m_lease, m_endpoint.CreateFault(Request, e));
                 }
             }
 
@@ -163,11 +164,11 @@ namespace Opc.Ua
             {
                 if (ServiceResult.IsBad(error))
                 {
-                    m_vts.SetResult(m_endpoint.CreateFault(Request, new ServiceResultException(error)));
+                    m_vts.TrySetResult(m_lease, m_endpoint.CreateFault(Request, new ServiceResultException(error)));
                 }
                 else
                 {
-                    m_vts.SetResult(response!);
+                    m_vts.TrySetResult(m_lease, response!);
                 }
             }
 
@@ -207,6 +208,12 @@ namespace Opc.Ua
 
             private readonly EndpointBase m_endpoint;
             private readonly ServiceResponsePooledValueTaskSource m_vts;
+
+            /// <summary>
+            /// The rental of <see cref="m_vts"/> owned by this request. Completions
+            /// are accepted once and never reach a later owner of the pooled source.
+            /// </summary>
+            private readonly int m_lease;
             private readonly RequestParkSink? m_parkSink;
             private CancellationToken m_transportCancellationToken;
         }

src/Opc.Ua.Core/Types/Utils/ValueTask/ServiceResponsePooledValueTaskSource.cs

@@ -37,19 +37,28 @@ namespace Opc.Ua
     /// <summary>
     /// A pooled value task source for IServiceResponse.
     /// </summary>
+    /// <remarks>
+    /// The pool is shared by every server in the process, so a completion that arrives
+    /// after the source was returned to the pool would answer whichever request rented it
+    /// next. Each rental therefore carries a lease and only the first completion with the
+    /// current lease is accepted.
+    /// </remarks>
     internal sealed class ServiceResponsePooledValueTaskSource : IValueTaskSource<IServiceResponse>, IValueTaskSource
     {
-        private static readonly ObjectPool<ServiceResponsePooledValueTaskSource> s_pool =
-            new(() => new ServiceResponsePooledValueTaskSource(), 1024);
+        private static readonly ObjectPool<ServiceResponsePooledValueTaskSource> s_pool = CreatePool(1024);
 
+        private readonly ObjectPool<ServiceResponsePooledValueTaskSource> m_pool;
         private readonly ManualResetValueTaskSource<IServiceResponse> m_source;
         private int m_resultRetrieved;
+        private int m_lease;
+        private int m_pendingLease;
 
         /// <summary>
         /// Private constructor to enforce pooling.
         /// </summary>
-        private ServiceResponsePooledValueTaskSource()
+        private ServiceResponsePooledValueTaskSource(ObjectPool<ServiceResponsePooledValueTaskSource> pool)
         {
+            m_pool = pool;
             m_source = new ManualResetValueTaskSource<IServiceResponse>();
         }
 
@@ -58,11 +67,34 @@ namespace Opc.Ua
         /// </summary>
         public static ServiceResponsePooledValueTaskSource Create()
         {
-            ServiceResponsePooledValueTaskSource source = s_pool.Get();
+            return Create(s_pool);
+        }
+
+        /// <summary>
+        /// Creates or gets an instance from the specified pool.
+        /// </summary>
+        internal static ServiceResponsePooledValueTaskSource Create(
+            ObjectPool<ServiceResponsePooledValueTaskSource> pool)
+        {
+            ServiceResponsePooledValueTaskSource source = pool.Get();
+            // Leases never repeat: a source is retired before its lease would wrap.
+            int lease = ++source.m_lease;
             source.m_resultRetrieved = 0;
+            Volatile.Write(ref source.m_pendingLease, lease);
             return source;
         }
 
+        /// <summary>
+        /// Creates an empty pool for sources.
+        /// </summary>
+        internal static ObjectPool<ServiceResponsePooledValueTaskSource> CreatePool(int maxSize)
+        {
+            ObjectPool<ServiceResponsePooledValueTaskSource>? pool = null;
+            pool = new ObjectPool<ServiceResponsePooledValueTaskSource>(
+                () => new ServiceResponsePooledValueTaskSource(pool!), maxSize);
+            return pool;
+        }
+
         /// <summary>
         /// Returns the object to the pool.
         /// </summary>
@@ -71,10 +103,19 @@ namespace Opc.Ua
             if (Interlocked.CompareExchange(ref m_resultRetrieved, 1, 0) == 0)
             {
                 m_source.Reset();
-                s_pool.Return(this);
+                if (m_lease != int.MaxValue)
+                {
+                    m_pool.Return(this);
+                }
             }
         }
 
+        /// <summary>
+        /// The lease of the current rental, captured by the owner when the
+        /// source is created and passed back to <see cref="TrySetResult"/>.
+        /// </summary>
+        public int Lease => m_lease;
+
         /// <summary>
         /// The value task to await.
         /// </summary>
@@ -89,19 +130,20 @@ namespace Opc.Ua
         public short Version => m_source.Version;
 
         /// <summary>
-        /// Set the result of the task.
+        /// Completes the rental identified by <paramref name="lease"/>.
         /// </summary>
-        public void SetResult(IServiceResponse result)
+        /// <returns>
+        /// <c>false</c> if that rental was already completed or the source was
+        /// recycled to another owner; the result is then discarded.
+        /// </returns>
+        public bool TrySetResult(int lease, IServiceResponse result)
         {
+            if (Interlocked.CompareExchange(ref m_pendingLease, 0, lease) != lease)
+            {
+                return false;
+            }
             m_source.SetResult(result);
-        }
-
-        /// <summary>
-        /// Set an exception for the task.
-        /// </summary>
-        public void SetException(Exception error)
-        {
-            m_source.SetException(error);
+            return true;
         }
 
         /// <inheritdoc/>

tests/Opc.Ua.Core.Tests/Stack/Server/EndpointIncomingRequestTests.cs

@@ -231,6 +231,32 @@ namespace Opc.Ua.Core.Tests.Stack.Server
             Assert.That(response.ResponseHeader.ServiceResult, Is.EqualTo(StatusCodes.BadUnexpectedError));
         }
 
+        /// <summary>
+        /// A scheduler that completes the request and then throws must not complete it twice (#4729).
+        /// </summary>
+        [Test]
+        public async Task ProcessAsync_ScheduleCompletesThenThrows_KeepsFirstCompletionAsync()
+        {
+            using var server = new TestServer();
+            var expected = new ReadResponse();
+            server.OnScheduleIncomingRequest = (r) =>
+            {
+                r.OperationCompleted(expected, ServiceResult.Good);
+                throw new ObjectDisposedException("queue");
+            };
+
+            var endpoint = new TestEndpointBase(server);
+            var req = new ReadRequest { RequestHeader = new RequestHeader() };
+            var ctx = new SecureChannelContext("1", new EndpointDescription(), RequestEncoding.Binary);
+
+            object incoming = endpoint.CreateIncomingRequest(req, ctx);
+            IServiceResponse response = await TestEndpointBase.ProcessAsyncLocal(incoming).ConfigureAwait(false);
+
+            Assert.That(response, Is.SameAs(expected));
+            Assert.DoesNotThrow(() => TestEndpointBase.OperationCompletedLocal(
+                incoming, null!, new ServiceResult(StatusCodes.BadServerHalted)));
+        }
+
         [Test]
         public async Task CallAsync_FindsServiceAndInvokesItAsync()
         {

tests/Opc.Ua.Core.Tests/Types/Utils/ServiceResponsePooledValueTaskSourceTests.cs

@@ -0,0 +1,112 @@
+/* ========================================================================
+ * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
+ *
+ * OPC Foundation MIT License 1.00
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following
+ * conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+ * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ *
+ * The complete license agreement can be found here:
+ * http://opcfoundation.org/License/MIT/1.00/
+ * ======================================================================*/
+
+using System.Reflection;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace Opc.Ua.Core.Tests.Types.UtilsTests
+{
+    /// <summary>
+    /// Tests that a pooled service response source accepts exactly one completion per rental.
+    /// </summary>
+    [TestFixture]
+    [Category("Utils")]
+    [Parallelizable]
+    public sealed class ServiceResponsePooledValueTaskSourceTests
+    {
+        [Test]
+        public async Task SecondCompletionOfSameRentalIsIgnoredAsync()
+        {
+            ObjectPool<ServiceResponsePooledValueTaskSource> pool =
+                ServiceResponsePooledValueTaskSource.CreatePool(1);
+            ServiceResponsePooledValueTaskSource source = ServiceResponsePooledValueTaskSource.Create(pool);
+            int lease = source.Lease;
+            ValueTask<IServiceResponse> task = source.Task;
+            var first = new ReadResponse();
+
+            Assert.That(source.TrySetResult(lease, first), Is.True);
+            Assert.That(source.TrySetResult(lease, new BrowseResponse()), Is.False);
+
+            Assert.That(await task.ConfigureAwait(false), Is.SameAs(first));
+        }
+
+        /// <summary>
+        /// A completion that arrives after the source was recycled must not answer the next owner (#4729).
+        /// </summary>
+        [Test]
+        public async Task LateCompletionDoesNotReachNextRentalAsync()
+        {
+            ObjectPool<ServiceResponsePooledValueTaskSource> pool =
+                ServiceResponsePooledValueTaskSource.CreatePool(1);
+
+            ServiceResponsePooledValueTaskSource first = ServiceResponsePooledValueTaskSource.Create(pool);
+            int staleLease = first.Lease;
+            ValueTask<IServiceResponse> firstTask = first.Task;
+            Assert.That(first.TrySetResult(staleLease, new CreateSessionResponse()), Is.True);
+            Assert.That(await firstTask.ConfigureAwait(false), Is.InstanceOf<CreateSessionResponse>());
+
+            ServiceResponsePooledValueTaskSource second = ServiceResponsePooledValueTaskSource.Create(pool);
+            Assert.That(second, Is.SameAs(first), "The pool should hand out the recycled source.");
+            ValueTask<IServiceResponse> secondTask = second.Task;
+
+            Assert.That(first.TrySetResult(staleLease, new CreateSessionResponse()), Is.False);
+            Assert.That(secondTask.IsCompleted, Is.False);
+
+            var expected = new ActivateSessionResponse();
+            Assert.That(second.TrySetResult(second.Lease, expected), Is.True);
+            Assert.That(await secondTask.ConfigureAwait(false), Is.SameAs(expected));
+        }
+
+        /// <summary>
+        /// A source whose lease reached int.MaxValue is not pooled again, so leases never repeat.
+        /// </summary>
+        [Test]
+        public async Task SourceIsRetiredBeforeLeaseWrapsAsync()
+        {
+            ObjectPool<ServiceResponsePooledValueTaskSource> pool =
+                ServiceResponsePooledValueTaskSource.CreatePool(1);
+            ServiceResponsePooledValueTaskSource source = ServiceResponsePooledValueTaskSource.Create(pool);
+            ValueTask<IServiceResponse> task = source.Task;
+            Assert.That(source.TrySetResult(source.Lease, new ReadResponse()), Is.True);
+            await task.ConfigureAwait(false);
+
+            typeof(ServiceResponsePooledValueTaskSource)
+                .GetField("m_lease", BindingFlags.NonPublic | BindingFlags.Instance)!
+                .SetValue(source, int.MaxValue - 1);
+            Assert.That(ServiceResponsePooledValueTaskSource.Create(pool), Is.SameAs(source));
+            Assert.That(source.Lease, Is.EqualTo(int.MaxValue));
+            task = source.Task;
+            Assert.That(source.TrySetResult(source.Lease, new ReadResponse()), Is.True);
+            await task.ConfigureAwait(false);
+
+            Assert.That(ServiceResponsePooledValueTaskSource.Create(pool), Is.Not.SameAs(source));
+        }
+    }
+}