refactor: simplify after code review

- ProtobufReader: extract shared ReadLengthDelimited() helper, eliminating
  duplicated bounds-check logic across ReadBytes/ReadString/ReadEmbeddedMessage
- ProtobufReader: add ReadBytesAsSpan() to avoid byte[] allocation for trace ID
- OtlpReceiver: replace unbounded ConcurrentBag<Task> with ConcurrentDictionary
  that self-cleans via ContinueWith — prevents memory leak under sustained load
- OtlpReceiver: track forwarding tasks so DisposeAsync awaits them too
- OtlpReceiver: extract MaxPortBindingAttempts constant
- AspireFixture: cache HttpMessageHandler across CreateHttpClient calls to avoid
  allocating new SocketsHttpHandler per client

TUnit.Aspire/AspireFixture.cs

@@ -28,6 +28,7 @@ public class AspireFixture<TAppHost> : IAsyncInitializer, IAsyncDisposable
 {
     private DistributedApplication? _app;
     private Telemetry.OtlpReceiver? _otlpReceiver;
+    private HttpMessageHandler? _httpHandler;
 
     /// <summary>
     /// The running Aspire distributed application.
@@ -46,7 +47,7 @@ public class AspireFixture<TAppHost> : IAsyncInitializer, IAsyncDisposable
     /// <returns>An <see cref="HttpClient"/> configured to connect to the resource with baggage propagation.</returns>
     public HttpClient CreateHttpClient(string resourceName, string? endpointName = null)
     {
-        var handler = new Http.TUnitBaggagePropagationHandler
+        _httpHandler ??= new Http.TUnitBaggagePropagationHandler
         {
             InnerHandler = new SocketsHttpHandler
             {
@@ -55,7 +56,7 @@ public HttpClient CreateHttpClient(string resourceName, string? endpointName = n
             },
         };
 
-        return new HttpClient(handler)
+        return new HttpClient(_httpHandler, disposeHandler: false)
         {
             BaseAddress = App.GetEndpoint(resourceName, endpointName),
         };
@@ -346,6 +347,9 @@ public virtual async ValueTask DisposeAsync()
             _otlpReceiver = null;
         }
 
+        _httpHandler?.Dispose();
+        _httpHandler = null;
+
         GC.SuppressFinalize(this);
     }
 

TUnit.Aspire/Telemetry/OtlpLogParser.cs

@@ -144,10 +144,10 @@ private static void ParseScopeLogs(ProtobufReader reader, string resourceName, L
                     break;
 
                 case 9 when wireType == WireType.LengthDelimited:
-                    var bytes = reader.ReadBytes();
-                    if (bytes.Length == 16)
+                    var traceBytes = reader.ReadBytesAsSpan();
+                    if (traceBytes.Length == 16)
                     {
-                        traceId = Convert.ToHexString(bytes);
+                        traceId = Convert.ToHexString(traceBytes);
                     }
 
                     break;
@@ -285,49 +285,39 @@ public ulong ReadVarint()
         return result;
     }
 
-    public byte[] ReadBytes()
+    public ReadOnlySpan<byte> ReadBytesAsSpan()
     {
-        var length = (int)ReadVarint();
-
-        if ((uint)length > (uint)_data.Length)
-        {
-            throw new InvalidOperationException(
-                $"Protobuf length-delimited field declares {length} bytes but only {_data.Length} remain.");
-        }
+        return ReadLengthDelimited();
+    }
 
-        var result = _data[..length].ToArray();
-        _data = _data[length..];
-        return result;
+    public byte[] ReadBytes()
+    {
+        return ReadLengthDelimited().ToArray();
     }
 
     public string ReadString()
     {
-        var length = (int)ReadVarint();
-
-        if ((uint)length > (uint)_data.Length)
-        {
-            throw new InvalidOperationException(
-                $"Protobuf string field declares {length} bytes but only {_data.Length} remain.");
-        }
-
-        var result = Encoding.UTF8.GetString(_data[..length]);
-        _data = _data[length..];
-        return result;
+        return Encoding.UTF8.GetString(ReadLengthDelimited());
     }
 
     public ProtobufReader ReadEmbeddedMessage()
+    {
+        return new ProtobufReader(ReadLengthDelimited());
+    }
+
+    private ReadOnlySpan<byte> ReadLengthDelimited()
     {
         var length = (int)ReadVarint();
 
         if ((uint)length > (uint)_data.Length)
         {
             throw new InvalidOperationException(
-                $"Protobuf embedded message declares {length} bytes but only {_data.Length} remain.");
+                $"Protobuf length-delimited field declares {length} bytes but only {_data.Length} remain.");
         }
 
-        var embedded = new ProtobufReader(_data[..length]);
+        var result = _data[..length];
         _data = _data[length..];
-        return embedded;
+        return result;
     }
 
     public void Skip(WireType wireType)

TUnit.Aspire/Telemetry/OtlpReceiver.cs

@@ -24,12 +24,15 @@ namespace TUnit.Aspire.Telemetry;
 /// </remarks>
 internal sealed class OtlpReceiver : IAsyncDisposable
 {
+    private const int MaxPortBindingAttempts = 10;
+
     private readonly HttpListener _listener;
     private readonly CancellationTokenSource _cts = new();
-    private readonly ConcurrentBag<Task> _inflightRequests = [];
+    private readonly ConcurrentDictionary<int, Task> _inflightTasks = new();
     private readonly HttpClient? _forwardingClient;
     private readonly string? _upstreamEndpoint;
     private Task? _listenTask;
+    private int _taskIdCounter;
 
     /// <summary>
     /// The port the receiver is listening on.
@@ -81,11 +84,17 @@ private async Task ListenLoop()
             }
 
             // Process each request without blocking the listen loop
-            var task = Task.Run(() => ProcessRequestAsync(context));
-            _inflightRequests.Add(task);
+            TrackTask(Task.Run(() => ProcessRequestAsync(context)));
         }
     }
 
+    private void TrackTask(Task task)
+    {
+        var id = Interlocked.Increment(ref _taskIdCounter);
+        _inflightTasks[id] = task;
+        task.ContinueWith(_ => _inflightTasks.TryRemove(id, out Task? _), TaskContinuationOptions.ExecuteSynchronously);
+    }
+
     private async Task ProcessRequestAsync(HttpListenerContext context)
     {
         if (_cts.IsCancellationRequested)
@@ -122,10 +131,10 @@ private async Task ProcessRequestAsync(HttpListenerContext context)
                 ProcessLogs(body);
             }
 
-            // Forward to upstream if configured (fire-and-forget)
+            // Forward to upstream if configured
             if (_upstreamEndpoint is not null && _forwardingClient is not null)
             {
-                _ = ForwardAsync(path, body, request.ContentType);
+                TrackTask(ForwardAsync(path, body, request.ContentType));
             }
 
             // Return 200 OK with empty protobuf response
@@ -238,11 +247,11 @@ public async ValueTask DisposeAsync()
             }
         }
 
-        // Wait for any in-flight request processing to complete so we don't
+        // Wait for any in-flight request/forwarding tasks to complete so we don't
         // access TraceRegistry or TestContext after they've been torn down.
         try
         {
-            await Task.WhenAll(_inflightRequests).ConfigureAwait(false);
+            await Task.WhenAll(_inflightTasks.Values).ConfigureAwait(false);
         }
         catch
         {
@@ -259,7 +268,7 @@ public async ValueTask DisposeAsync()
     /// </summary>
     private static (HttpListener Listener, int Port) CreateListener()
     {
-        for (var attempt = 0; attempt < 10; attempt++)
+        for (var attempt = 0; attempt < MaxPortBindingAttempts; attempt++)
         {
             var port = FindFreePort();
             var listener = new HttpListener();
@@ -276,7 +285,7 @@ private static (HttpListener Listener, int Port) CreateListener()
             }
         }
 
-        throw new InvalidOperationException("Could not bind OTLP listener after 10 attempts.");
+        throw new InvalidOperationException($"Could not bind OTLP listener after {MaxPortBindingAttempts} attempts.");
     }
 
     private static int FindFreePort()