TCP listener: complete requests without an async state machine per request (R7) (#4788)

Roadmap item R7 of the o6 benchmark gap analysis (master `5b03e9afa` profile). `TcpTransportListener.OnRequestReceivedAsync` was an `async void` method. Every request allocated an `AsyncStateMachineBox` when it awaited the request processing, which almost always completes later on a queue worker. In the profile the box was about 4 % of the server's on-CPU samples (`GetStateMachineBox`) and 4 % of the allocated bytes. The listener now completes a request without an async state machine. The handling of every other outcome is unchanged: - failures; - sends that wait for the channel gate; - closed channels; - the retry on the session's new channel. No wire, configuration or public API change. ## Results Paired A/B against master `af601ebca` (= `5b03e9afa` + #4778). The builds alternate per round, with the order rotated each round, on the same machine. The table shows medians of 5 rounds. The machine was shared with other work; allocated bytes per request are deterministic. - **o6read** uses the native open62541 client workers of the o6-automation benchmark, in the o6 2.0.1 configuration: `AsyncCustomNodeManager`, Server GC, queue = clients × outstanding + 3. - **"4 cores"** pins the server to 4 cores (`--server-affinity 0xF`, `DOTNET_PROCESSOR_COUNT=4`). - **The other rows** use the harness's .NET client, 20 sessions unless noted. | Scenario | req/s | Server CPU µs/req | Alloc B/req | |---|---|---|---| | o6read 10×100, 4 cores | 130,300 → 130,200 | 30.3 → 29.8 (−2 %) | 4,706 → 4,554 (**−152 B**) | | o6read 10×100 encrypted, 4 cores | 116,300 → 118,500 (+2 %) | 33.4 → 32.8 (−2 %) | 4,747 → 4,606 (**−141 B**) | | o6read 10×100, 24 cores | 344,500 → 340,200 (−1 %) | 53.6 → 54.1 (+1 %) | 4,710 → 4,509 (**−4 %**) | | o6read 10×1000, 24 cores | 310,800 → 306,800 (−1 %) | 55.6 → 57.7 (+4 %, noise) | 4,719 → 4,530 (**−4 %**) | | read --nodes 1, 20 sessions | 189,500 → 200,800 (+6 %) | 49.7 → 49.0 (−1 %) | 4,624 → 4,413 (**−5 %**) | | read --nodes 1, 200 sessions | 174,500 → 183,200 (+5 %) | 50.2 → 46.7 (−7 %) | 4,510 → 4,322 (**−4 %**) | | browse | 36,490 → 36,220 | 179 → 174 (−3 %) | 39,004 → 38,796 | | sub (5 × 100 items, 100 ms) | 94,020 → 93,530 notif/s | 8.2 → 7.5 (rounds 4.5–11.2) | 145 → 143 | **Effect:** - **Allocation.** Every request through the TCP listener allocates 140–210 B less (−3 to −5 %). - **CPU.** CPU per request falls by about 2 % on 4 cores. On 24 cores it is within the spread of the rounds, where waiting dominates. ### Profile cpu and alloc traces, 24 cores, 10×100, quiet runs at 433–466k req/s. The traces were taken with this PR and #4787 (R3) together; the frames below belong to this PR only. | Frame | master | this PR | |---|---|---| | `GetStateMachineBox` of `OnRequestReceivedAsync`, exclusive | 4.0 % of on-CPU samples | 0.03 % | | `OnRequestReceivedAsync` state machine box | 4.1 % of allocated bytes | 0 | | `PendingRequest` (pool rent and return) | – | not visible; the time under `PendingRequest.OnProcessed` (1.9 %) is the channel's send-order lock | ## Changes - **`OnRequestReceived`** starts the processing synchronously. A request still processing is awaited by a pooled `PendingRequest`. It keeps the request's state in fields and continues with a cached delegate. The response value task source is pooled already, so a request allocates nothing for this in steady state. The continuation flows the execution context, as the `await` did. - **Synchronous completion.** When the processing succeeded and the channel queued the response without waiting, the request completes synchronously. The request and response are returned to their pools and the pending-request count is released. - **`CompleteRequestAsync`** is an `async void` method with the previous body, and handles every other outcome: - a processing failure, thrown synchronously or later (logged, answered with a service fault); - a send that waits for the channel gate; - a send on a closed channel (quiet log) and its retry on the session's new channel. - **Kestrel TCP and WebSocket** use a different handler (no service fault, no pending-request scope, no request reuse). They are left unchanged. ## Testing - **`TcpTransportListenerRequestCompletionTests` (new)**, with a loopback listener and a UA-TCP client: - 1200 concurrent requests on 4 channels each get their own response. The requests complete synchronously, through a task, or later through a value task source on another thread, so the pooled continuations are reused; - a processing failure, thrown synchronously, returned as a faulted task, or set later, is logged as an error once and answered with a service fault, and the channel keeps working; - a callback value task source that throws when its state is read, the continuation is registered or the result is read fails only that request: logged, answered with a service fault, channel keeps working (these time out without the fix in ee81c83f2); - a cancelled request is answered with the endpoint's fault; - a response completed after the client closed its channel is dropped with the quiet log and without an error, and the listener keeps serving; - no unobserved task exceptions. The same tests pass on master. - **Local, net10.0.** With this PR alone: Opc.Ua.Core.Tests Stack.Tcp + Stack.Transport, 847 passed, 0 failed. With both PRs applied (all 0 failed): - Opc.Ua.Core.Tests (Stack.Server, Stack.Tcp, Stack.Transport): 2753 passed; - Opc.Ua.Server.Tests: 8750 passed; - Opc.Ua.Sessions.Tests: 857 passed; - Opc.Ua.Redundancy.Server.Tests: 884 passed; Opc.Ua.Redundancy.Client.Tests: 134 passed; - Opc.Ua.Stress.Tests Contract + Integration: 45 passed. net48 builds. `dotnet format whitespace --verify-no-changes` passes on the changed files. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 05:11:13 +08:00 · 2026-10-15 12:53:19 +02:00
parent 19a97da52e
commit 95850b7e55
2 changed files with 881 additions and 41 deletions
@@ -0,0 +1,604 @@
+/* ========================================================================
+ * Copyright (c) 2005-2026 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;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+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;
+using Opc.Ua.Bindings;
+using Opc.Ua.Security.Certificates;
+
+namespace Opc.Ua.Core.Tests.Stack.Transport
+{
+    /// <summary>
+    /// Loopback tests of how the <see cref="TcpTransportListener"/> completes a request: the
+    /// response is sent for a request that completes synchronously or later, and a request
+    /// whose processing fails is logged and answered with a service fault.
+    /// </summary>
+    [TestFixture]
+    [Category("TcpTransport")]
+    [SetCulture("en-us")]
+    [SetUICulture("en-us")]
+    [NonParallelizable]
+    public sealed class TcpTransportListenerRequestCompletionTests
+    {
+        [SetUp]
+        public void SetUp()
+        {
+            m_unobserved = new ConcurrentQueue<Exception>();
+            TaskScheduler.UnobservedTaskException += OnUnobservedTaskException;
+        }
+
+        [TearDown]
+        public void TearDown()
+        {
+            GC.Collect();
+            GC.WaitForPendingFinalizers();
+            GC.Collect();
+            TaskScheduler.UnobservedTaskException -= OnUnobservedTaskException;
+            Assert.That(m_unobserved, Is.Empty, "A request completion left an unobserved exception.");
+        }
+
+        /// <summary>
+        /// Concurrent requests that complete synchronously, through a task, or later through a
+        /// pooled value task source on another thread each get their own response, also while
+        /// the listener reuses its pending-request continuations.
+        /// </summary>
+        [Test]
+        [CancelAfter(60000)]
+        public async Task EveryRequestGetsItsOwnResponseAcrossCompletionKindsAsync()
+        {
+            const int kChannels = 4;
+            const int kRequestsPerChannel = 300;
+            var callback = new ScriptedCallback(request =>
+            {
+                uint handle = request.RequestHeader.RequestHandle;
+                IServiceResponse response = Respond(request);
+                switch (handle % 3)
+                {
+                    case 0:
+                        return new ValueTask<IServiceResponse>(response);
+                    case 1:
+                        return new ValueTask<IServiceResponse>(Task.Run(async () =>
+                        {
+                            await Task.Yield();
+                            return response;
+                        }));
+                    default:
+                        var source = new ManualResetValueTaskSource<IServiceResponse>();
+                        _ = Task.Run(async () =>
+                        {
+                            await Task.Yield();
+                            source.SetResult(response);
+                        });
+                        return source.Task;
+                }
+            });
+            await using Loopback loopback = await Loopback.StartAsync(callback).ConfigureAwait(false);
+
+            var clients = new UaSCUaBinaryTransportChannel[kChannels];
+            try
+            {
+                for (int ii = 0; ii < kChannels; ii++)
+                {
+                    clients[ii] = await loopback.ConnectAsync().ConfigureAwait(false);
+                }
+                var requests = new List<Task<(uint Sent, uint Received)>>();
+                for (int ii = 0; ii < kChannels * kRequestsPerChannel; ii++)
+                {
+                    UaSCUaBinaryTransportChannel client = clients[ii % kChannels];
+                    uint handle = (uint)ii + 1;
+                    requests.Add(Task.Run(async () =>
+                    {
+                        IServiceResponse response = await client.SendRequestAsync(
+                            CreateReadRequest(handle), CancellationToken.None).ConfigureAwait(false);
+                        return (handle, response.ResponseHeader.RequestHandle);
+                    }));
+                }
+                (uint Sent, uint Received)[] results = await Task.WhenAll(requests).ConfigureAwait(false);
+
+                Assert.That(results.Where(r => r.Sent != r.Received), Is.Empty);
+                Assert.That(callback.RequestCount, Is.EqualTo(kChannels * kRequestsPerChannel));
+                Assert.That(loopback.Logs.Where(l => l.Level >= LogLevel.Error), Is.Empty);
+            }
+            finally
+            {
+                foreach (UaSCUaBinaryTransportChannel client in clients)
+                {
+                    client?.Dispose();
+                }
+            }
+        }
+
+        /// <summary>
+        /// A request whose processing fails, synchronously, through a faulted task or later
+        /// through a value task source, is logged as an error and answered with a service fault,
+        /// and the listener keeps serving the channel.
+        /// </summary>
+        [TestCase(FailureKind.SynchronousThrow)]
+        [TestCase(FailureKind.FaultedTask)]
+        [TestCase(FailureKind.LaterFault)]
+        [CancelAfter(30000)]
+        public async Task ProcessingFailureIsLoggedAndAnsweredWithServiceFaultAsync(FailureKind kind)
+        {
+            var callback = new ScriptedCallback(request =>
+            {
+                if (request.RequestHeader.RequestHandle != 1)
+                {
+                    return new ValueTask<IServiceResponse>(Respond(request));
+                }
+                var error = new ServiceResultException(StatusCodes.BadNodeIdUnknown, "Injected failure.");
+                switch (kind)
+                {
+                    case FailureKind.SynchronousThrow:
+                        throw error;
+                    case FailureKind.FaultedTask:
+                        return new ValueTask<IServiceResponse>(Task.FromException<IServiceResponse>(error));
+                    default:
+                        var source = new ManualResetValueTaskSource<IServiceResponse>();
+                        _ = Task.Run(async () =>
+                        {
+                            await Task.Yield();
+                            source.SetException(error);
+                        });
+                        return source.Task;
+                }
+            });
+            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.BadNodeIdUnknown));
+            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 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>
+        [Test]
+        [CancelAfter(30000)]
+        public async Task CancelledProcessingIsAnsweredWithServiceFaultAsync()
+        {
+            var callback = new ScriptedCallback(received =>
+            {
+                var source = new ManualResetValueTaskSource<IServiceResponse>();
+                _ = Task.Run(async () =>
+                {
+                    await Task.Yield();
+                    source.SetException(new OperationCanceledException());
+                });
+                return source.Task;
+            });
+            await using Loopback loopback = await Loopback.StartAsync(callback).ConfigureAwait(false);
+            using UaSCUaBinaryTransportChannel client = await loopback.ConnectAsync().ConfigureAwait(false);
+            ReadRequest request = CreateReadRequest(1);
+            StatusCode expected = EndpointBase.CreateFault(
+                loopback.Telemetry.CreateLogger<TcpTransportListenerRequestCompletionTests>(),
+                request, new OperationCanceledException()).ResponseHeader.ServiceResult;
+
+            IServiceResponse fault = await client.SendRequestAsync(request, CancellationToken.None)
+                .ConfigureAwait(false);
+
+            Assert.That(fault, Is.InstanceOf<ServiceFault>());
+            Assert.That(fault.ResponseHeader.ServiceResult, Is.EqualTo(expected));
+        }
+
+        /// <summary>
+        /// A response that completes after the client closed its channel cannot be sent: the
+        /// listener logs that quietly, sends no fault, and keeps serving other channels.
+        /// </summary>
+        [Test]
+        [CancelAfter(30000)]
+        public async Task ResponseAfterChannelClosedIsDroppedQuietlyAsync()
+        {
+            var processing = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var source = new ManualResetValueTaskSource<IServiceResponse>();
+            var callback = new ScriptedCallback(request =>
+            {
+                if (request.RequestHeader.RequestHandle != 1)
+                {
+                    return new ValueTask<IServiceResponse>(Respond(request));
+                }
+                processing.TrySetResult(true);
+                return source.Task;
+            });
+            await using Loopback loopback = await Loopback.StartAsync(callback).ConfigureAwait(false);
+            UaSCUaBinaryTransportChannel closing = await loopback.ConnectAsync().ConfigureAwait(false);
+            Task<IServiceResponse> abandoned = closing.SendRequestAsync(CreateReadRequest(1), CancellationToken.None)
+                .AsTask();
+            await processing.Task.ConfigureAwait(false);
+            await closing.CloseAsync(CancellationToken.None).ConfigureAwait(false);
+            closing.Dispose();
+            Assert.CatchAsync(async () => await abandoned.ConfigureAwait(false));
+
+            source.SetResult(Respond(CreateReadRequest(1)));
+
+            Assert.That(await WaitForAsync(() => loopback.Logs.Any(l =>
+                l.EventId == CoreEventIds.TcpTransportListener + 21))
+                .ConfigureAwait(false), Is.True, "The dropped response was not logged.");
+            Assert.That(loopback.Logs.Where(l => l.Level >= LogLevel.Error), Is.Empty);
+            using UaSCUaBinaryTransportChannel other = await loopback.ConnectAsync().ConfigureAwait(false);
+            IServiceResponse response = await other.SendRequestAsync(CreateReadRequest(2), CancellationToken.None)
+                .ConfigureAwait(false);
+            Assert.That(response.ResponseHeader.RequestHandle, Is.EqualTo(2));
+        }
+
+        public enum FailureKind
+        {
+            SynchronousThrow,
+            FaultedTask,
+            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);
+        }
+
+        private static ReadResponse Respond(IServiceRequest request)
+        {
+            return new ReadResponse
+            {
+                ResponseHeader = new ResponseHeader
+                {
+                    RequestHandle = request.RequestHeader.RequestHandle,
+                    ServiceResult = StatusCodes.Good
+                }
+            };
+        }
+
+        private static ReadRequest CreateReadRequest(uint handle)
+        {
+            return new ReadRequest
+            {
+                RequestHeader = new RequestHeader { RequestHandle = handle, TimeoutHint = 10000 },
+                NodesToRead = [new ReadValueId { NodeId = new NodeId(1001, 1), AttributeId = Attributes.Value }]
+            };
+        }
+
+        private static async Task<bool> WaitForAsync(Func<bool> condition)
+        {
+            DateTime deadline = DateTime.UtcNow.AddSeconds(10);
+            while (DateTime.UtcNow < deadline)
+            {
+                if (condition())
+                {
+                    return true;
+                }
+                await Task.Delay(20).ConfigureAwait(false);
+            }
+            return condition();
+        }
+
+        private ConcurrentQueue<Exception> m_unobserved = null!;
+
+        /// <summary>
+        /// A listener on a free loopback port with a log that the tests read.
+        /// </summary>
+        private sealed class Loopback : IAsyncDisposable
+        {
+            private Loopback(DefaultTelemetry telemetry, CapturingLoggerProvider logs, Uri endpointUrl)
+            {
+                Telemetry = telemetry;
+                m_logs = logs;
+                m_endpointUrl = endpointUrl;
+                m_listener = new TcpTransportListener(telemetry);
+            }
+
+            public ITelemetryContext Telemetry { get; }
+
+            public IReadOnlyCollection<LogEntry> Logs => m_logs.Entries.ToArray();
+
+            public static async Task<Loopback> StartAsync(ITransportListenerCallback callback)
+            {
+                var logs = new CapturingLoggerProvider();
+                var telemetry = (DefaultTelemetry)DefaultTelemetry.Create(
+                    builder => builder.SetMinimumLevel(LogLevel.Debug).AddProvider(logs));
+                var loopback = new Loopback(
+                    telemetry, logs, new Uri($"opc.tcp://127.0.0.1:{GetFreeTcpPort()}"));
+                var certificateRegistry = new Mock<ICertificateRegistry>();
+                certificateRegistry
+                    .Setup(r => r.AcquireApplicationCertificateBySecurityPolicy(It.IsAny<string>()))
+                    .Returns((CertificateEntry)null!);
+                await loopback.m_listener.OpenAsync(
+                    loopback.m_endpointUrl,
+                    new TransportListenerSettings
+                    {
+                        Descriptions = [loopback.Endpoint],
+                        Configuration = CreateConfiguration(),
+                        ServerCertificates = certificateRegistry.Object,
+                        NamespaceUris = new NamespaceTable(),
+                        Factory = EncodeableFactory.Create(),
+                        MaxChannelCount = 10
+                    },
+                    callback,
+                    CancellationToken.None).ConfigureAwait(false);
+                return loopback;
+            }
+
+            public async Task<UaSCUaBinaryTransportChannel> ConnectAsync()
+            {
+                var channel = new UaSCUaBinaryTransportChannel(new TcpByteTransportFactory(Telemetry), Telemetry)
+                {
+                    OperationTimeout = 10000
+                };
+                try
+                {
+                    await channel.OpenAsync(
+                        m_endpointUrl,
+                        new TransportChannelSettings
+                        {
+                            Description = Endpoint,
+                            Configuration = CreateConfiguration(),
+                            NamespaceUris = new NamespaceTable(),
+                            Factory = EncodeableFactory.Create()
+                        },
+                        CancellationToken.None).ConfigureAwait(false);
+                    return channel;
+                }
+                catch
+                {
+                    channel.Dispose();
+                    throw;
+                }
+            }
+
+            public async ValueTask DisposeAsync()
+            {
+                await m_listener.CloseAsync(CancellationToken.None).ConfigureAwait(false);
+                await m_listener.DisposeAsync().ConfigureAwait(false);
+                ((IDisposable)Telemetry).Dispose();
+            }
+
+            private EndpointDescription Endpoint => new()
+            {
+                EndpointUrl = m_endpointUrl.ToString(),
+                SecurityMode = MessageSecurityMode.None,
+                SecurityPolicyUri = SecurityPolicies.None,
+                TransportProfileUri = Profiles.UaTcpTransport
+            };
+
+            private static EndpointConfiguration CreateConfiguration()
+            {
+                var configuration = EndpointConfiguration.Create();
+                configuration.OperationTimeout = 10000;
+                configuration.ChannelLifetime = 60000;
+                configuration.SecurityTokenLifetime = 60000;
+                return configuration;
+            }
+
+            private static int GetFreeTcpPort()
+            {
+                var listener = new TcpListener(IPAddress.Loopback, 0);
+                listener.Start();
+                try
+                {
+                    return ((IPEndPoint)listener.LocalEndpoint).Port;
+                }
+                finally
+                {
+                    listener.Stop();
+                }
+            }
+
+            private readonly CapturingLoggerProvider m_logs;
+            private readonly Uri m_endpointUrl;
+            private readonly TcpTransportListener m_listener;
+        }
+
+        private readonly record struct LogEntry(LogLevel Level, int EventId, string Message);
+
+        private sealed class CapturingLoggerProvider : ILoggerProvider
+        {
+            public ConcurrentQueue<LogEntry> Entries { get; } = new();
+
+            public ILogger CreateLogger(string categoryName)
+            {
+                return new CapturingLogger(Entries);
+            }
+
+            public void Dispose()
+            {
+            }
+
+            private sealed class CapturingLogger : ILogger
+            {
+                public CapturingLogger(ConcurrentQueue<LogEntry> entries)
+                {
+                    m_entries = entries;
+                }
+
+                public IDisposable BeginScope<TState>(TState state)
+                    where TState : notnull
+                {
+                    return null!;
+                }
+
+                public bool IsEnabled(LogLevel logLevel)
+                {
+                    return true;
+                }
+
+                public void Log<TState>(
+                    LogLevel logLevel,
+                    EventId eventId,
+                    TState state,
+                    Exception? exception,
+                    Func<TState, Exception?, string> formatter)
+                {
+                    m_entries.Enqueue(new LogEntry(logLevel, eventId.Id, formatter(state, exception)));
+                }
+
+                private readonly ConcurrentQueue<LogEntry> m_entries;
+            }
+        }
+
+        /// <summary>
+        /// Answers each request with the result of a script.
+        /// </summary>
+        private sealed class ScriptedCallback : ITransportListenerCallback
+        {
+            public ScriptedCallback(Func<IServiceRequest, ValueTask<IServiceResponse>> script)
+            {
+                m_script = script;
+            }
+
+            public int RequestCount => Volatile.Read(ref m_requestCount);
+
+            public ValueTask<IServiceResponse> ProcessRequestAsync(
+                SecureChannelContext secureChannelContext,
+                IServiceRequest request,
+                CancellationToken cancellationToken = default)
+            {
+                Interlocked.Increment(ref m_requestCount);
+                return m_script(request);
+            }
+
+            public bool TryGetSecureChannelIdForAuthenticationToken(NodeId authenticationToken, out uint channelId)
+            {
+                channelId = 0;
+                return false;
+            }
+
+            public void ReportAuditOpenSecureChannelEvent(
+                string globalChannelId,
+                EndpointDescription endpointDescription,
+                OpenSecureChannelRequest request,
+                Certificate clientCertificate,
+                Exception exception)
+            {
+            }
+
+            public void ReportAuditCloseSecureChannelEvent(string globalChannelId, Exception exception)
+            {
+            }
+
+            public void ReportAuditCertificateEvent(Certificate clientCertificate, Exception exception)
+            {
+            }
+
+            private readonly Func<IServiceRequest, ValueTask<IServiceResponse>> m_script;
+            private int m_requestCount;
+        }
+    }
+}