Many fixes

Author: st0o0

src/TurboHttp.Benchmarks/StreamingThroughputBenchmarks.cs

@@ -1,11 +1,11 @@
 using Akka.Actor;
 using Akka.Configuration;
 using BenchmarkDotNet.Attributes;
-using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Options;
 using TurboHttp.Benchmarks.Internal;
+using TurboHttp.Internal;
 
 namespace TurboHttp.Benchmarks;
 
@@ -33,7 +33,8 @@ public class StreamingThroughputBenchmarks
     private static readonly Lock ServerLock = new();
     private static int _serverRefCount;
 
-    private static readonly Config BenchHocon = ConfigurationFactory.Empty;
+    private static readonly Config BenchHocon =
+        TurboHttpDispatchers.CreateConfig(TurboClientOptions.DefaultMaxEndpointSubstreams);
 
     private ServiceProvider? _turboProvider;
     private ITurboHttpClient? _turboClient;
@@ -62,7 +63,7 @@ public class StreamingThroughputBenchmarks
         : _server!.Http11Port;
 
     [GlobalSetup]
-    public void GlobalSetup()
+    public async Task GlobalSetup()
     {
         lock (ServerLock)
         {
@@ -81,8 +82,8 @@ public void GlobalSetup()
 
         SetupHttpClient();
 
-        WarmupTurbo().GetAwaiter().GetResult();
-        WarmupHttpClient().GetAwaiter().GetResult();
+        await WarmupTurbo();
+        await WarmupHttpClient();
     }
 
     [GlobalCleanup]
@@ -143,14 +144,12 @@ public async Task TurboHttp_StreamRequests()
 
         // Fire all requests as fast as the channel accepts them
 
-        var tasks = new Task[count];
         for (var i = 0; i < count; i++)
         {
             var request = new HttpRequestMessage(HttpMethod.Get, "/benchmark/simple");
-            tasks[i] = writer.WriteAsync(request).AsTask();
+            await writer.WriteAsync(request);
         }
 
-        await Task.WhenAll(tasks);
 
         // Drain all responses — use manual loop instead of ReadAllAsync to avoid
         // IAsyncEnumerator disposal issues when breaking out early
@@ -260,4 +259,4 @@ private sealed class FixedOptionsFactory(TurboClientOptions options) : IOptionsF
     {
         public TurboClientOptions Create(string name) => options;
     }
-}
+}

src/TurboHttp.Benchmarks/TurboHttpComparativeBenchmarks.cs

@@ -5,6 +5,7 @@
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Options;
 using TurboHttp.Benchmarks.Internal;
+using TurboHttp.Internal;
 
 namespace TurboHttp.Benchmarks;
 
@@ -16,21 +17,21 @@ namespace TurboHttp.Benchmarks;
 /// </summary>
 internal sealed class ClientHelper : IAsyncDisposable
 {
-    private static readonly Config BenchHocon = ConfigurationFactory.Empty;
+    private static readonly Config BenchHocon =
+        TurboHttpDispatchers.CreateConfig(TurboClientOptions.DefaultMaxEndpointSubstreams);
 
     private readonly ServiceProvider _provider;
-    private readonly ITurboHttpClient _client;
     private readonly ActorSystem _system;
 
     private ClientHelper(ServiceProvider provider, ITurboHttpClient client, ActorSystem system)
     {
         _provider = provider;
-        _client = client;
+        Client = client;
         _system = system;
     }
 
     /// <summary>The configured <see cref="ITurboHttpClient"/> instance.</summary>
-    public ITurboHttpClient Client => _client;
+    public ITurboHttpClient Client { get; }
 
     /// <summary>
     /// Creates a new <see cref="ClientHelper"/> with a fully configured TurboHttp client.
@@ -80,18 +81,18 @@ public static ClientHelper CreateClient(int port, Version version)
     public async ValueTask DisposeAsync()
     {
         // Signal pipeline to drain
-        _client.Requests.TryComplete();
+        Client.Requests.TryComplete();
 
         try
         {
-            await _client.Responses.Completion.WaitAsync(TimeSpan.FromSeconds(5));
+            await Client.Responses.Completion.WaitAsync(TimeSpan.FromSeconds(5));
         }
         catch
         {
             // Pipeline may complete with an error during shutdown — that is fine.
         }
 
-        _client.Dispose();
+        Client.Dispose();
 
         // Terminate the ActorSystem to stop all PinnedDispatcher threads.
         // Without this, each BDN parameter combination leaks ~50–100 OS threads, causing

src/TurboHttp.IntegrationTests/LoggingBridgeSpec.cs

@@ -7,6 +7,7 @@
 using Microsoft.Extensions.Logging;
 using TurboHttp.Diagnostics;
 using TurboHttp.IntegrationTests.Shared;
+using TurboHttp.Internal;
 
 namespace TurboHttp.IntegrationTests;
 
@@ -98,8 +99,10 @@ private ITurboHttpClient BuildClientViaUserDI(bool withTurboTrace = false)
         {
             var loggerFactory = sp.GetRequiredService<ILoggerFactory>();
             var diSetup = DependencyResolverSetup.Create(sp);
+            var dispatcherConfig = TurboHttpDispatchers.CreateConfig(
+                TurboClientOptions.DefaultMaxEndpointSubstreams);
             var setup = BootstrapSetup.Create()
-                .WithConfig(LoggingHocon)
+                .WithConfig(LoggingHocon.WithFallback(dispatcherConfig))
                 .And(diSetup)
                 .And(new LoggerFactorySetup(loggerFactory));
             return ActorSystem.Create("turbohttp-bridge-test", setup);

src/TurboHttp.IntegrationTests/Shared/ActorSystemFixture.cs

@@ -3,6 +3,7 @@
 using Akka.Configuration;
 using Akka.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection;
+using TurboHttp.Internal;
 
 namespace TurboHttp.IntegrationTests.Shared;
 
@@ -19,7 +20,9 @@ public ValueTask InitializeAsync()
     {
         var services = new ServiceCollection();
         var diSetup = DependencyResolverSetup.Create(services.BuildServiceProvider());
-        var bootstrap = BootstrapSetup.Create();
+        var dispatcherConfig = TurboHttpDispatchers.CreateConfig(
+            TurboClientOptions.DefaultMaxEndpointSubstreams);
+        var bootstrap = BootstrapSetup.Create().WithConfig(dispatcherConfig);
 
         var setup = bootstrap.And(diSetup);
         System = ActorSystem.Create($"turbohttp-shared-{Guid.NewGuid()}", setup);

src/TurboHttp.IntegrationTests/Shared/ClientHelper.cs

@@ -6,6 +6,7 @@
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using TurboHttp.Internal;
 
 namespace TurboHttp.IntegrationTests.Shared;
 
@@ -65,10 +66,14 @@ public static ClientHelper CreateClient(
             // Create an ActorSystem with DependencyResolver so that Servus.Akka
             // ResolveActor<T> works inside TurboClientStreamManager.
             var diSetup = DependencyResolverSetup.Create(services.BuildServiceProvider());
+            var dispatcherConfig = TurboHttpDispatchers.CreateConfig(
+                TurboClientOptions.DefaultMaxEndpointSubstreams);
             var bootstrap = BootstrapSetup.Create();
 
             if (loggerFactory is not null)
-                bootstrap = bootstrap.WithConfig(LoggingHocon);
+                bootstrap = bootstrap.WithConfig(LoggingHocon.WithFallback(dispatcherConfig));
+            else
+                bootstrap = bootstrap.WithConfig(dispatcherConfig);
 
             var setup = loggerFactory is not null
                 ? bootstrap.And(diSetup).And(new LoggerFactorySetup(loggerFactory))

src/TurboHttp.StreamTests/Dispatchers/DispatcherConfigurationSpec.cs

@@ -0,0 +1,160 @@
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Streams;
+using TurboHttp.Internal;
+using TurboHttp.Transport.Connection;
+
+namespace TurboHttp.StreamTests.Dispatchers;
+
+/// <summary>
+/// Verifies that TurboHttp's custom Akka.NET dispatchers are correctly configured,
+/// resolvable from ActorSystem, and applied to the right actors.
+/// </summary>
+public sealed class DispatcherConfigurationSpec : IAsyncDisposable
+{
+    private readonly ActorSystem _system;
+
+    public DispatcherConfigurationSpec()
+    {
+        var config = TurboHttpDispatchers.CreateConfig(TurboClientOptions.DefaultMaxEndpointSubstreams);
+        _system = ActorSystem.Create("dispatcher-spec-" + Guid.NewGuid(), config);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _system.Terminate();
+    }
+
+    [Fact(Timeout = 5000)]
+    public void CreateConfig_should_produce_ForkJoinDispatcher_for_io_dispatcher()
+    {
+        var config = TurboHttpDispatchers.CreateConfig(256);
+        var type = config.GetString("akka.actor.turbohttp-io-dispatcher.type");
+
+        Assert.Equal("ForkJoinDispatcher", type);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void CreateConfig_should_produce_ForkJoinDispatcher_for_stream_dispatcher()
+    {
+        var config = TurboHttpDispatchers.CreateConfig(256);
+        var type = config.GetString("akka.actor.turbohttp-stream-dispatcher.type");
+
+        Assert.Equal("ForkJoinDispatcher", type);
+    }
+
+    [Theory(Timeout = 5000)]
+    [InlineData(32u)]
+    [InlineData(64u)]
+    [InlineData(128u)]
+    [InlineData(256u)]
+    [InlineData(512u)]
+    [InlineData(1024u)]
+    public void CreateConfig_should_clamp_stream_thread_count(uint maxEndpointSubstreams)
+    {
+        var config = TurboHttpDispatchers.CreateConfig(maxEndpointSubstreams);
+        var threadCount = config.GetInt(
+            "akka.actor.turbohttp-stream-dispatcher.dedicated-thread-pool.thread-count");
+
+        Assert.InRange(threadCount, Environment.ProcessorCount, 64);
+    }
+
+    [Theory(Timeout = 5000)]
+    [InlineData(32u)]
+    [InlineData(256u)]
+    [InlineData(1024u)]
+    public void CreateConfig_should_clamp_io_thread_count(uint maxEndpointSubstreams)
+    {
+        var config = TurboHttpDispatchers.CreateConfig(maxEndpointSubstreams);
+        var threadCount = config.GetInt(
+            "akka.actor.turbohttp-io-dispatcher.dedicated-thread-pool.thread-count");
+
+        Assert.InRange(threadCount, 4, 16);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void IoDispatcher_should_be_resolvable_from_ActorSystem()
+    {
+        Assert.True(
+            _system.Dispatchers.HasDispatcher(TurboHttpDispatchers.IoDispatcher));
+    }
+
+    [Fact(Timeout = 5000)]
+    public void StreamDispatcher_should_be_resolvable_from_ActorSystem()
+    {
+        Assert.True(
+            _system.Dispatchers.HasDispatcher(TurboHttpDispatchers.StreamDispatcher));
+    }
+
+    [Fact(Timeout = 5000)]
+    public void WithIoDispatcher_should_apply_when_available()
+    {
+        var props = Props.Create(() => new ConnectionManagerActor(TimeSpan.FromMinutes(5)));
+
+        Assert.Equal(TurboHttpDispatchers.IoDispatcher, props.WithIoDispatcher(_system).Dispatcher);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void WithIoDispatcher_should_fall_back_to_default_when_missing()
+    {
+        using var bareSystem = ActorSystem.Create("bare-" + Guid.NewGuid());
+        var props = Props.Create(() => new ConnectionManagerActor(TimeSpan.FromMinutes(5)));
+
+        // Should NOT have the custom dispatcher — falls back to default
+        Assert.NotEqual(TurboHttpDispatchers.IoDispatcher, props.WithIoDispatcher(bareSystem).Dispatcher);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void WithStreamDispatcher_should_apply_when_available()
+    {
+        var props = Props.Create(() => new ConnectionManagerActor(TimeSpan.FromMinutes(5)));
+
+        Assert.Equal(TurboHttpDispatchers.StreamDispatcher, props.WithStreamDispatcher(_system).Dispatcher);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void WithStreamDispatcher_should_fall_back_to_default_when_missing()
+    {
+        using var bareSystem = ActorSystem.Create("bare-" + Guid.NewGuid());
+        var settings = ActorMaterializerSettings.Create(bareSystem);
+
+        // Should NOT have the custom dispatcher — falls back to default
+        Assert.NotEqual(TurboHttpDispatchers.StreamDispatcher, settings.WithStreamDispatcher(bareSystem).Dispatcher);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void CreateConfig_should_use_background_threads()
+    {
+        var config = TurboHttpDispatchers.CreateConfig(256);
+
+        Assert.Equal("background",
+            config.GetString("akka.actor.turbohttp-io-dispatcher.dedicated-thread-pool.threadtype"));
+        Assert.Equal("background",
+            config.GetString("akka.actor.turbohttp-stream-dispatcher.dedicated-thread-pool.threadtype"));
+    }
+
+    [Fact(Timeout = 5000)]
+    public void CreateConfig_should_set_expected_throughput_values()
+    {
+        var config = TurboHttpDispatchers.CreateConfig(256);
+
+        Assert.Equal(32, config.GetInt("akka.actor.turbohttp-io-dispatcher.throughput"));
+        Assert.Equal(64, config.GetInt("akka.actor.turbohttp-stream-dispatcher.throughput"));
+    }
+
+    [Fact(Timeout = 5000)]
+    public void CreateConfig_should_scale_stream_threads_with_substreams()
+    {
+        // With high MaxEndpointSubstreams, thread count should be higher than with low
+        var configLow = TurboHttpDispatchers.CreateConfig(32);
+        var configHigh = TurboHttpDispatchers.CreateConfig(512);
+
+        var threadsLow = configLow.GetInt(
+            "akka.actor.turbohttp-stream-dispatcher.dedicated-thread-pool.thread-count");
+        var threadsHigh = configHigh.GetInt(
+            "akka.actor.turbohttp-stream-dispatcher.dedicated-thread-pool.thread-count");
+
+        Assert.True(threadsHigh >= threadsLow,
+            $"Expected high substreams ({threadsHigh}) >= low substreams ({threadsLow})");
+    }
+}

src/TurboHttp.StreamTests/Http10/Http10DecompressionPipelineSpec.cs

@@ -110,7 +110,7 @@ public async Task Http10DecompressionPipeline_should_remove_content_encoding_hea
 
     [Fact(Timeout = 10_000)]
     [Trait("RFC", "RFC1945-10.3")]
-    public async Task Http10DecompressionPipeline_should_update_content_length_when_gzip_decompressed()
+    public async Task Http10DecompressionPipeline_should_produce_correct_body_length_when_gzip_decompressed()
     {
         var request = new HttpRequestMessage(HttpMethod.Get, "http://example.com/gzip-len")
         {
@@ -126,7 +126,8 @@ public async Task Http10DecompressionPipeline_should_update_content_length_when_
             request,
             () => rawResponse);
 
-        Assert.Equal(original.Length, response.Content.Headers.ContentLength);
+        var body = await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken);
+        Assert.Equal(original.Length, body.Length);
     }
 
     // x-gzip decompression

src/TurboHttp.StreamTests/Http11/Chunking/Http11ChunkedTransferSpec.cs

@@ -79,9 +79,17 @@ public async Task Http11ChunkedTransfer_should_concatenate_three_equal_chunks_wh
     [Trait("RFC", "RFC9112-7.1")]
     public async Task Http11ChunkedTransfer_should_end_stream_when_zero_length_terminator()
     {
-        var responses = await DecodeAllAsync(
-            "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" +
-            "5\r\nhello\r\n0\r\n\r\n");
+        // Use Take(1) + Sink.Seq instead of bare Sink.Seq to avoid waiting
+        // for upstream completion — the decoder stage keeps the stream open
+        // for potential pipelined responses.
+        var responses = await Source.From(new[]
+            {
+                Chunk("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" +
+                      "5\r\nhello\r\n0\r\n\r\n")
+            })
+            .Via(Flow.FromGraph(new Http11DecoderStage()))
+            .Take(1)
+            .RunWith(Sink.Seq<HttpResponseMessage>(), Materializer);
 
         Assert.Single(responses);
         var body = await responses[0].Content.ReadAsStringAsync(TestContext.Current.CancellationToken);

src/TurboHttp.StreamTests/Http11/Decoding/Http11DecoderSpec.cs

@@ -70,8 +70,11 @@ public async Task Http11Decoder_should_decode_pipelined_responses_when_two_respo
             Chunk("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\nHTTP/1.1 201 Created\r\nContent-Length: 0\r\n\r\n")
         ]);
 
+        // Take(2) ensures the stream completes after collecting both responses
+        // instead of waiting for upstream completion (which Sink.Seq requires).
         var responses = await source
             .Via(Flow.FromGraph(new Http11DecoderStage()))
+            .Take(2)
             .RunWith(Sink.Seq<HttpResponseMessage>(), Materializer);
 
         Assert.Equal(2, responses.Count);