Fix transport buffer pooling, limiting, and in-process capture (#3994)

# Description

Fixes transport-buffer pool amplification when
`TransportQuotas.MaxBufferSize` is an exact power of two, adds
injectable buffer-manager policies and a process-wide memory limiter,
and fixes in-process pcap capture of maximum-size UASC chunks.

`BufferManager` previously reserved an ownership cookie after the usable
data in every build. A 65,536-byte limit therefore requested 65,537
bytes from `ArrayPool<byte>.Shared`, selecting the 131,072-byte bucket.
The compatibility `BufferManager` is now a facade over `IBufferManager`;
Release defaults to `FastBufferManager`, Debug defaults to
`CookieBufferManager`, and explicit `TRACK_MEMORY` builds use
`TracingBufferManager`.

`IBufferManagerFactory` and `BufferManagerFactoryOptions` are registered
through dependency injection and wired through raw TCP, HTTPS/WSS, and
Kestrel channel/listener creation while preserving existing direct
constructors. Applications can replace the factory or explicitly select
fast, cookie, or tracing behavior.

`LimitingBufferManager` wraps any implementation and uses a
factory-shared `BufferManagerMemoryLimiter` to cap actual outstanding
bytes across all managers created by that factory. It blocks synchronous
rents without holding manager locks, supports cancellation on transport
I/O paths, accounts for actual pool bucket sizes, rejects impossible or
synchronous re-entrant rents safely, and protects accounting from
failed, foreign, and duplicate returns.

The client and accepted-server TCP transports also retained raw quota
values. The fix carries bucket-safe limits into transport creation and
reconnects, while preserving the active connection’s negotiated receive
limit. Configurable transports receive updated limits during server
reconnect, and every channel independently rejects and faults/reconnects
on chunks above the negotiated limit.

Maximum-size UASC chunks exceeded the synthetic IPv4 total-length field
once TCP/IP headers were added. The capture path now segments those
chunks into sequence-aware synthetic TCP packets sized for both IPv4 and
the pcap snap length. Every capturing transport receives a stable
process-unique synthetic flow identity that is preserved in the full TCP
tuple and independent sequence state. Capture queuing remains
non-blocking and queues or drops complete UASC chunks atomically;
sequence space is reserved before bounded frame admission so dropped
chunks remain visible as TCP gaps, while packetization stays in the
single writer. Key-material snapshots use a separate prioritized
unbounded queue and are drained and disposed reliably during shutdown.

This change includes:

- `IBufferManager`, `IBufferManagerFactory`, factory options, and
compatibility facade;
- fast, cookie, allocation-tracing, and memory-limiting implementations;
- shared DI factory wiring for raw TCP, HTTPS/WSS, and Kestrel
transports;
- Debug/Release/TRACK_MEMORY implementation selection;
- cancellation-aware transport rents and no-loss in-process receive
behavior;
- deterministic ownership, limiter, bucket-size, reconnect,
oversized-chunk, cancellation, and DI tests;
- explicit subscription diagnostics using `ArrayPoolEventSource`;
- pcap framing, sequence-space, atomic-queue, and reassembly regression
tests;
- updated benchmark and dependency-injection documentation.

Measured with 255 and 257 Publish queues plus a slow-consumer scenario,
configured values of both 65,535 and 65,536 used only 65,536-byte
arrays. Every observed rent had a matching return, outstanding buffers
returned to zero, and no Gen 2 collections occurred. The 500-session
baseline also passed for `opc.tcp`, `https`, and `opc.https`, with all
sessions receiving notifications.

## Related Issues

- Fixes #3988

## Checklist

- [x] 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.
- [x] I have addressed **all** PR feedback received.

Author: marcschier

Docs/Benchmarks.md

@@ -427,6 +427,16 @@ The numbers below were measured on an **Intel Xeon W-2235 (6 physical cores / 12
 
 (*) These figures are hardware- and load-dependent and are directional; they improve on dedicated hardware where the client and server run on separate machines. With `ServerConfiguration.DecoupleHeldPublishRequests` on (the default) and `MaxRequestThreadCount` sized to the active establishment concurrency (~200) rather than to the session count, this 6-core machine cleanly established every session and delivered 100 % of notifications up to **~4000** concurrent sessions. At 10000 it tips into a secure-channel connect storm (repeated handshake aborts/retries and tens of thousands of channel attempts for the target) and does not establish the full set within the connect budget - so on this hardware establishment, not notification delivery, is the ceiling; more cores push it higher. When sessions establish but do not receive notifications the last column instead reads `No (N drops)`, where `N` is the number of sessions that did not receive notifications within the window. See [Server Session Scalability](ServerScalability.md) for the code-referenced analysis and the built-in controls for degrading gracefully under load.
 
+### Subscription transport buffer pooling
+
+`BufferManager` previously stored a one-byte ownership cookie after the usable transport data in all builds. Configuring `TransportQuotas.MaxBufferSize` to an exact power of two such as 65,536 could therefore call `ArrayPool<byte>.Rent(65,537)`, selecting the 131,072-byte bucket. The cookie is now enabled in Debug builds (and explicit `TRACK_MEMORY` diagnostic builds) but removed from normal Release builds; the secure-channel normalization still keeps data plus any active diagnostic cookie in the same bucket.
+
+Channels now normalize their advertised send and receive limits to the largest protocol-safe value that keeps the data plus cookie in the same `ArrayPool` bucket (while preserving the OPC UA 8,192-byte minimum), and both TCP transport creation paths receive that normalized limit. A configured value of either 65,535 or 65,536 consequently uses 65,536-byte arrays instead of 131,072-byte arrays.
+
+The explicit net10.0 harness `SubscriptionArrayPoolDiagnosticsLoadTests` (`Tests/Opc.Ua.Sessions.Tests/SubscriptionArrayPoolDiagnosticsLoadTests.cs`) measures real subscription traffic with 255 and 257 session Publish queues (immediately below and above the parallel Publish threshold) plus a slow-consumer case. Across both configured buffer sizes, all measured scenarios used only the 65,536-byte bucket, rents and returns balanced exactly after quiescence, outstanding buffers returned to zero, peak outstanding buffers stayed between 528 and 536, and the measurement windows triggered no Gen 2 collections. The 500-session regression baseline also completed for `opc.tcp`, `https`, and `opc.https`, with all 500 sessions receiving notifications.
+
+The net10.0 in-process short `BufferManagerBenchmarks` run completed 256 `BufferManager.TakeBuffer`/`ReturnBuffer` pairs in 56.18 us versus 52.52 us for direct `ArrayPool<byte>.Shared` use, with no managed allocation reported for either pooled path. These short-run timings are directional; the subscription harness is the authoritative ownership and bucket-size check.
+
 ### Sizing and configuration
 
 * `MaxSessionCount` (default 100) caps the concurrent open sessions; size `MaxChannelCount` (one channel per session) and `MaxSubscriptionCount` above the target session count.

Docs/DependencyInjection.md

@@ -85,12 +85,14 @@ lifetime, certificate setup, and Ctrl+C / SIGTERM handling.
 
 ## Root: `services.AddOpcUa()`
 
-`services.AddOpcUa()` is the only entry point. It does two things:
+`services.AddOpcUa()` is the only entry point. It does three things:
 
 1. Registers `ITelemetryContext` as a `ServiceProviderTelemetryContext`
    singleton (via `TryAddSingleton`, so any prior user registration
    wins).
-2. Returns an `IOpcUaBuilder` whose `.Services` property exposes the
+2. Registers the default `IBufferManagerFactory` and
+   `BufferManagerFactoryOptions` singletons (also via `TryAddSingleton`).
+3. Returns an `IOpcUaBuilder` whose `.Services` property exposes the
    underlying `IServiceCollection` for advanced scenarios.
 
 ```csharp
@@ -114,6 +116,35 @@ services.AddOpcUa()
     .AddNodeManager<MyNodeManagerFactory>();
 ```
 
+### Buffer managers
+
+Transport listeners and channels resolve `IBufferManagerFactory` from dependency injection. The default factory selects `FastBufferManager` in Release builds, `CookieBufferManager` in Debug builds, and `TracingBufferManager` when the stack is compiled with `TRACK_MEMORY`.
+
+Register options before `AddOpcUa()` to select an implementation explicitly or apply a process-wide outstanding-buffer budget:
+
+```csharp
+services.AddSingleton(new BufferManagerFactoryOptions
+{
+    ImplementationKind = BufferManagerImplementationKind.Fast,
+    MaxOutstandingBytesPerProcess = 256L * 1024 * 1024
+});
+
+services.AddOpcUa()
+    .AddOpcTcpTransport()
+    .AddHttpsTransport();
+```
+
+When `MaxOutstandingBytesPerProcess` is positive, the singleton factory wraps every manager it creates with `LimitingBufferManager` and shares one `BufferManagerMemoryLimiter` across them. A synchronous rent blocks without holding a manager lock until another buffer is returned. A single rent whose conservative expected size exceeds the budget fails immediately instead of waiting forever.
+
+Applications can replace the complete policy by registering an `IBufferManagerFactory` before `AddOpcUa()`:
+
+```csharp
+services.AddSingleton<IBufferManagerFactory, MyBufferManagerFactory>();
+services.AddOpcUa();
+```
+
+The existing `new BufferManager(name, maxBufferSize, telemetry)` path remains available for direct creation. `BufferManager` is a compatibility facade over `IBufferManager`; `FastBufferManager`, `CookieBufferManager`, `TracingBufferManager`, and `LimitingBufferManager` can also be created directly for advanced scenarios.
+
 Discovery servers expose the same transport and reverse-connect shortcuts as
 the regular server builder:
 

Stack/Opc.Ua.Bindings.Https/DependencyInjection/OpcUaHttpsBuilderExtensions.cs

@@ -217,10 +217,14 @@ private static IOpcUaBuilder AddWssBindings(IOpcUaBuilder builder)
 
         private static void RegisterWssFactories(ITransportBindingRegistry registry, IServiceProvider provider)
         {
+            IBufferManagerFactory bufferManagerFactory =
+                provider.GetRequiredService<IBufferManagerFactory>();
             RegisterHttpsListener(registry, provider, new WssTransportListenerFactory());
             RegisterHttpsListener(registry, provider, new OpcWssTransportListenerFactory());
-            registry.RegisterChannelFactory(new WssTransportChannelFactory());
-            registry.RegisterChannelFactory(new OpcWssTransportChannelFactory());
+            registry.RegisterChannelFactory(
+                new WssTransportChannelFactory(bufferManagerFactory));
+            registry.RegisterChannelFactory(
+                new OpcWssTransportChannelFactory(bufferManagerFactory));
             registry.RegisterChannelFactory(new WssJsonTransportChannelFactory());
         }
 
@@ -229,6 +233,9 @@ private static void RegisterHttpsListener(
             IServiceProvider provider,
             HttpsServiceHost factory)
         {
+            factory.BufferManagerFactory =
+                provider.GetService<IBufferManagerFactory>() ??
+                DefaultBufferManagerFactory.Instance;
             foreach (IHttpsListenerStartupContributor contributor in
                 provider.GetServices<IHttpsListenerStartupContributor>())
             {

Stack/Opc.Ua.Bindings.Https/DependencyInjection/OpcUaKestrelTcpBuilderExtensions.cs

@@ -63,8 +63,14 @@ public static IOpcUaBuilder AddKestrelOpcTcpTransport(this IOpcUaBuilder builder
             }
 
             builder.Services.AddTransportBindingRegistry();
-            builder.Services.AddSingleton<ITransportBindingConfigurator>(
-                new TransportBindingConfigurator(registry => registry.RegisterListenerFactory(new KestrelTcpTransportListenerFactory())));
+            builder.Services.AddSingleton<ITransportBindingConfigurator>(provider =>
+            {
+                IBufferManagerFactory bufferManagerFactory =
+                    provider.GetRequiredService<IBufferManagerFactory>();
+                return new TransportBindingConfigurator(
+                    registry => registry.RegisterListenerFactory(
+                        new KestrelTcpTransportListenerFactory(bufferManagerFactory)));
+            });
             return builder;
         }
     }

Stack/Opc.Ua.Bindings.Https/Https/HttpsServiceHost.cs

@@ -59,6 +59,12 @@ public abstract class HttpsServiceHost : ITransportListenerFactory
         public IList<IHttpsListenerStartupContributor> StartupContributors { get; }
             = [];
 
+        /// <summary>
+        /// Gets or sets the factory used to create listener buffer managers.
+        /// </summary>
+        internal IBufferManagerFactory BufferManagerFactory { get; set; } =
+            DefaultBufferManagerFactory.Instance;
+
         /// <summary>
         /// The OPC UA <c>TransportProfileUri</c> reported on the
         /// <see cref="EndpointDescription.TransportProfileUri"/> emitted for

Stack/Opc.Ua.Bindings.Https/Https/HttpsTransportListener.cs

@@ -76,7 +76,11 @@ public class HttpsTransportListenerFactory : HttpsServiceHost
         /// <returns>The transport listener.</returns>
         public override ITransportListener Create(ITelemetryContext telemetry)
         {
-            return new HttpsTransportListener(Utils.UriSchemeHttps, telemetry, [.. StartupContributors]);
+            return new HttpsTransportListener(
+                Utils.UriSchemeHttps,
+                telemetry,
+                [.. StartupContributors],
+                BufferManagerFactory);
         }
     }
 
@@ -103,7 +107,11 @@ public class OpcHttpsTransportListenerFactory : HttpsServiceHost
         /// <returns>The transport listener.</returns>
         public override ITransportListener Create(ITelemetryContext telemetry)
         {
-            return new HttpsTransportListener(Utils.UriSchemeOpcHttps, telemetry, [.. StartupContributors]);
+            return new HttpsTransportListener(
+                Utils.UriSchemeOpcHttps,
+                telemetry,
+                [.. StartupContributors],
+                BufferManagerFactory);
         }
     }
 
@@ -129,7 +137,11 @@ public class WssTransportListenerFactory : HttpsServiceHost
         /// <inheritdoc/>
         public override ITransportListener Create(ITelemetryContext telemetry)
         {
-            return new HttpsTransportListener(Utils.UriSchemeWss, telemetry, [.. StartupContributors]);
+            return new HttpsTransportListener(
+                Utils.UriSchemeWss,
+                telemetry,
+                [.. StartupContributors],
+                BufferManagerFactory);
         }
     }
 
@@ -155,7 +167,11 @@ public class OpcWssTransportListenerFactory : HttpsServiceHost
         /// <inheritdoc/>
         public override ITransportListener Create(ITelemetryContext telemetry)
         {
-            return new HttpsTransportListener(Utils.UriSchemeOpcWss, telemetry, [.. StartupContributors]);
+            return new HttpsTransportListener(
+                Utils.UriSchemeOpcWss,
+                telemetry,
+                [.. StartupContributors],
+                BufferManagerFactory);
         }
     }
 
@@ -321,10 +337,12 @@ public class HttpsTransportListener : ITransportListener, ITransportListenerCert
         /// Initializes a new instance of the <see cref="HttpsTransportListener"/> class.
         /// </summary>
         public HttpsTransportListener(string uriScheme, ITelemetryContext telemetry)
+            : this(
+                uriScheme,
+                telemetry,
+                startupContributors: [],
+                DefaultBufferManagerFactory.Instance)
         {
-            UriScheme = uriScheme;
-            m_telemetry = telemetry;
-            m_logger = telemetry.CreateLogger<HttpsTransportListener>();
         }
 
         /// <summary>
@@ -338,9 +356,29 @@ internal HttpsTransportListener(
             string uriScheme,
             ITelemetryContext telemetry,
             IReadOnlyList<IHttpsListenerStartupContributor> startupContributors)
-            : this(uriScheme, telemetry)
+            : this(
+                uriScheme,
+                telemetry,
+                startupContributors,
+                DefaultBufferManagerFactory.Instance)
         {
+        }
+
+        /// <summary>
+        /// Initializes a listener with startup contributors and a buffer-manager factory.
+        /// </summary>
+        internal HttpsTransportListener(
+            string uriScheme,
+            ITelemetryContext telemetry,
+            IReadOnlyList<IHttpsListenerStartupContributor> startupContributors,
+            IBufferManagerFactory bufferManagerFactory)
+        {
+            UriScheme = uriScheme;
+            m_telemetry = telemetry;
+            m_logger = telemetry.CreateLogger<HttpsTransportListener>();
             StartupContributors = startupContributors;
+            m_bufferManagerFactory = bufferManagerFactory ??
+                throw new ArgumentNullException(nameof(bufferManagerFactory));
         }
 
         /// <summary>
@@ -550,9 +588,10 @@ public async ValueTask OpenAsync(
 
             // buffer manager used by the WSS path to rent send / receive chunks.
             m_bufferManager = new BufferManager(
-                "HttpsListener",
-                m_quotas.MaxBufferSize,
-                m_telemetry);
+                m_bufferManagerFactory.Create(
+                    "HttpsListener",
+                    m_quotas.MaxBufferSize,
+                    m_telemetry));
 
             // start the listener
             await StartAsync(ct).ConfigureAwait(false);
@@ -1786,7 +1825,8 @@ private async Task AcceptWebSocketJsonAsync(HttpContext context)
                 {
                     receiveBuffer ??= m_bufferManager.TakeBuffer(
                         m_quotas.MaxBufferSize,
-                        nameof(AcceptWebSocketJsonAsync));
+                        nameof(AcceptWebSocketJsonAsync),
+                        ct);
 
                     int totalRead = 0;
                     bool completed;
@@ -1998,7 +2038,8 @@ private async Task AcceptWebSocketOpenApiAsync(
                 {
                     receiveBuffer ??= m_bufferManager.TakeBuffer(
                         m_quotas.MaxBufferSize,
-                        nameof(AcceptWebSocketOpenApiAsync));
+                        nameof(AcceptWebSocketOpenApiAsync),
+                        ct);
 
                     int totalRead = 0;
                     bool completed;
@@ -2405,6 +2446,7 @@ private bool ValidateClientCertificate(
         private List<EndpointDescription> m_descriptions = null!;
         private ChannelQuotas m_quotas = null!;
         private BufferManager m_bufferManager = null!;
+        private readonly IBufferManagerFactory m_bufferManagerFactory;
         private ITransportListenerCallback? m_callback;
 #if NET8_0_OR_GREATER
         private IHost? m_host;

Stack/Opc.Ua.Bindings.Https/Https/WebSocketByteTransport.cs

@@ -120,7 +120,10 @@ public async ValueTask SendChunkAsync(BufferCollection buffers, CancellationToke
                 // WebSocket does not support a vectored send out of the box; concat the
                 // gathered segments into a single contiguous chunk for the frame.
                 int totalSize = buffers.TotalSize;
-                byte[] frame = m_bufferManager.TakeBuffer(totalSize, nameof(SendChunkAsync));
+                byte[] frame = m_bufferManager.TakeBuffer(
+                    totalSize,
+                    nameof(SendChunkAsync),
+                    ct);
                 try
                 {
                     int offset = 0;
@@ -167,7 +170,10 @@ await socket
         public async ValueTask<ArraySegment<byte>> ReceiveChunkAsync(CancellationToken ct)
         {
             WebSocket socket = RequireOpenSocket();
-            byte[] buffer = m_bufferManager.TakeBuffer(m_receiveBufferSize, nameof(ReceiveChunkAsync));
+            byte[] buffer = m_bufferManager.TakeBuffer(
+                m_receiveBufferSize,
+                nameof(ReceiveChunkAsync),
+                ct);
             int totalRead = 0;
             try
             {