TASK-032-003: Thread maxSubstreamsPerKey and slotQueueSize through extensions

Add `maxSubstreamsPerKey` (default 1) and `slotQueueSize` (default 64, renamed
from `queueSize`) parameters to `GroupByRequestKey<T, TMat>` extension and
`HostKeyMergeBack<T, TMat>`. Both are forwarded to `GroupByRequestKeyStage` so
callers can configure multi-slot routing without breaking existing call sites.

Author: st0o0

.maggus/features/feature_032.md

@@ -87,11 +87,11 @@ Two architectural improvements to TurboHttp's stream pipeline:
 **Parallel:** no
 
 **Acceptance Criteria:**
-- [ ] `GroupByRequestKey<T, TMat>` extension gains `maxSubstreamsPerKey = 1` parameter
-- [ ] `GroupByRequestKey<T, TMat>` extension gains `slotQueueSize = 64` parameter
-- [ ] `HostKeyMergeBack<T, TMat>` stores and passes both parameters to `GroupByRequestKeyStage`
-- [ ] Existing call sites compile without changes (defaults preserve old behavior)
-- [ ] Build succeeds with zero errors
+- [x] `GroupByRequestKey<T, TMat>` extension gains `maxSubstreamsPerKey = 1` parameter
+- [x] `GroupByRequestKey<T, TMat>` extension gains `slotQueueSize = 64` parameter
+- [x] `HostKeyMergeBack<T, TMat>` stores and passes both parameters to `GroupByRequestKeyStage`
+- [x] Existing call sites compile without changes (defaults preserve old behavior)
+- [x] Build succeeds with zero errors
 
 ---
 

src/TurboHttp.StreamTests/Streams/13_GroupByHostKeyQueueSizeTests.cs

@@ -54,7 +54,7 @@ public async Task Should_ControlQueueSize_When_ConstructorParameterSpecified()
 
         var flow = (Flow<HttpRequestMessage, HttpRequestMessage, NotUsed>)
             Flow.Create<HttpRequestMessage>()
-                .GroupByRequestKey(RequestEndpoint.FromRequest, maxSubstreams: 16, queueSize: 128)
+                .GroupByRequestKey(RequestEndpoint.FromRequest, maxSubstreams: 16, slotQueueSize: 128)
                 .MergeSubstreams();
 
         var results = await Source.From(requests)

src/TurboHttp/Streams/Stages/Internal/HostKeyGroupByExtensions.cs

@@ -19,9 +19,10 @@ public static SubFlow<T, TMat, Sink<T, TMat>> GroupByRequestKey<T, TMat>(
         this IFlow<T, TMat> flow,
         Func<T, RequestEndpoint> keyFunction,
         int maxSubstreams,
-        int queueSize = 64)
+        int slotQueueSize = 64,
+        int maxSubstreamsPerKey = 1)
     {
-        var mergeBack = new HostKeyMergeBack<T, TMat>(flow, keyFunction, maxSubstreams, queueSize);
+        var mergeBack = new HostKeyMergeBack<T, TMat>(flow, keyFunction, maxSubstreams, slotQueueSize, maxSubstreamsPerKey);
 
         // Flow.Create<T>() gives Flow<T,T,NotUsed>; cast is safe because callers always
         // start with a flow whose TMat is NotUsed (e.g. Flow.Create<HttpRequestMessage>()).

src/TurboHttp/Streams/Stages/Internal/HostKeyMergeBack.cs

@@ -15,15 +15,17 @@ internal sealed class HostKeyMergeBack<TIn, TMat> : IMergeBack<TIn, TMat>
     private readonly IFlow<TIn, TMat> _baseFlow;
     private readonly Func<TIn, RequestEndpoint> _keyFunction;
     private readonly int _maxSubstreams;
-    private readonly int _queueSize;
+    private readonly int _slotQueueSize;
+    private readonly int _maxSubstreamsPerKey;
 
     public HostKeyMergeBack(IFlow<TIn, TMat> baseFlow, Func<TIn, RequestEndpoint> keyFunction, int maxSubstreams,
-        int queueSize = 64)
+        int slotQueueSize = 64, int maxSubstreamsPerKey = 1)
     {
         _baseFlow = baseFlow;
         _keyFunction = keyFunction;
         _maxSubstreams = maxSubstreams;
-        _queueSize = queueSize;
+        _slotQueueSize = slotQueueSize;
+        _maxSubstreamsPerKey = maxSubstreamsPerKey;
     }
 
     // Called by SubFlowImpl.MergeSubstreamsWithParallelism(breadth).
@@ -36,7 +38,7 @@ public IFlow<TOut, TMat> Apply<TOut>(Flow<TIn, TOut, TMat> flow, int breadth)
             : breadth;
 
         return _baseFlow
-            .Via(new GroupByRequestKeyStage<TIn>(_keyFunction, _maxSubstreams, _queueSize))
+            .Via(new GroupByRequestKeyStage<TIn>(_keyFunction, _maxSubstreams, _slotQueueSize, _maxSubstreamsPerKey))
             .Via(Flow.Create<Source<TIn, NotUsed>>()
                 .Select(src => src.Via(flow)))
             .Via(new MergeSubstreamsStage<TOut>(effectiveBreadth));