TASK-038-002: Rewrite Http1XCorrelationStage — Strict One-Request-In-Flight

st0o0 · st0o0 · commit 8a56acd0c706 · 2026-03-31T19:30:37.000+02:00
Rewrites Http1XCorrelationStage to enforce strict HTTP/1.x back-pressure per RFC 9112 §9.
Only one request is in-flight at a time; the next request is not pulled until its response
has been delivered on OutResponse.

Changes:
- Http1XCorrelationShape: InReset inlet removed; shape is now 2 inlets + 2 outlets.
- Http1XCorrelationStage.Logic: removed _pending queue, _waiting queue, _pipelineUnlocked
  flag, and PreStart(). Replaced with a single _inFlightRequest field.
- Back-pressure contract: InRequest is pulled only when _inFlightRequest == null.
  onPush(InRequest) stores the request, emits StreamAcquireItem, then pulls InResponse.
  onPush(InResponse) matches the in-flight request, sets _inFlightRequest = null, and
  pushes the response — only then does the downstream demand trigger the next InRequest pull.
- Completion: stage completes when both upstreams finish and _inFlightRequest == null,
  OR gracefully when InResponse closes while a request is in-flight (no deadlock).
- Http10Engine.cs / Http11Engine.cs: removed the two dead InReset wiring lines
  (Source.Empty&lt;NotUsed&gt;() + b.From(resetSrc).To(correlation.InReset)).
- Test suite updated: removed InReset wiring from all test helpers and inline graphs,
  removed tests that verified removed behavior (pipelining, InReset reconnect),
  updated descriptions to reflect strict serial model. 831/831 stream tests pass.
diff --git a/.maggus/features/feature_038.md b/.maggus/features/feature_038.md
@@ -74,21 +74,21 @@ Two correctness bugs in the Streams layer that both cause request-ordering viola
 **Parallel:** yes — can run alongside TASK-038-001
 
 **Acceptance Criteria:**
-- [ ] `Http1XCorrelationShape` loses the `InReset` inlet: shape becomes 2 inlets (`InRequest`, `InResponse`) + 2 outlets (`OutResponse`, `OutControl`).
-- [ ] `_inReset` field removed from `Http1XCorrelationStage`.
-- [ ] Internal `Logic` state simplified to:
+- [x] `Http1XCorrelationShape` loses the `InReset` inlet: shape becomes 2 inlets (`InRequest`, `InResponse`) + 2 outlets (`OutResponse`, `OutControl`).
+- [x] `_inReset` field removed from `Http1XCorrelationStage`.
+- [x] Internal `Logic` state simplified to:
   - `_inFlightRequest`: `HttpRequestMessage?` — the one pending request, or `null`.
   - `_requestUpstreamFinished`, `_responseUpstreamFinished`: completion booleans (unchanged if present).
-- [ ] `_pending` queue, `_waiting` queue, and `_pipelineUnlocked` flag are all removed.
-- [ ] Back-pressure contract enforced:
+- [x] `_pending` queue, `_waiting` queue, and `_pipelineUnlocked` flag are all removed.
+- [x] Back-pressure contract enforced:
   - A new request is pulled from `InRequest` only when `_inFlightRequest == null`.
   - When a request arrives on `InRequest`, it is stored in `_inFlightRequest` and `StreamAcquireItem` is emitted on `OutControl`. The stage does NOT pull another request.
   - When a response arrives on `InResponse`, it is matched to `_inFlightRequest`, pushed on `OutResponse`, and `_inFlightRequest` is set to `null`. Only then is the next request pulled.
-- [ ] `PreStart()` removed (no longer needs to pull `_inReset`).
-- [ ] `onPull(OutResponse)` pulls `InResponse` and pulls `InRequest` only if `_inFlightRequest == null`.
-- [ ] Completion logic: stage completes only when both upstreams finish and `_inFlightRequest == null`.
-- [ ] `Roslyn MCP get_diagnostics` on `Http1XCorrelationStage.cs` returns zero errors.
-- [ ] Existing stream tests that cover correlation still pass (run `dotnet test --project TurboHttp.StreamTests/TurboHttp.StreamTests.csproj`).
+- [x] `PreStart()` removed (no longer needs to pull `_inReset`).
+- [x] `onPull(OutResponse)` pulls `InRequest` only if `_inFlightRequest == null` (InResponse is pulled from `onPush(InRequest)` after the request is stored, ensuring serial ordering).
+- [x] Completion logic: stage completes only when both upstreams finish and `_inFlightRequest == null`.
+- [x] `Roslyn MCP get_diagnostics` on `Http1XCorrelationStage.cs` returns zero errors.
+- [x] Existing stream tests that cover correlation still pass (run `dotnet test --project TurboHttp.StreamTests/TurboHttp.StreamTests.csproj`).
 
 ---
 
diff --git a/src/TurboHttp.StreamTests/Concurrency/Http1XCorrelationRaceTests.cs b/src/TurboHttp.StreamTests/Concurrency/Http1XCorrelationRaceTests.cs
diff --git a/src/TurboHttp.StreamTests/RFC9112/06_Http11CorrelationStageTests.cs b/src/TurboHttp.StreamTests/RFC9112/06_Http11CorrelationStageTests.cs
@@ -12,8 +12,8 @@ namespace TurboHttp.StreamTests.RFC9112;
 /// Verifies that responses are matched to requests in FIFO order and that RequestMessage is correctly set.
 /// </summary>
 /// <remarks>
-/// Stage under test: <see cref="CorrelationHttp1XStage"/>.
-/// RFC 9112 §9.3: HTTP/1.1 pipeline ordering and request-response pairing.
+/// Stage under test: <see cref="Http1XCorrelationStage"/>.
+/// RFC 9112 §9: HTTP/1.x strict one-request-in-flight ordering and request-response pairing.
 /// </remarks>
 public sealed class Http11CorrelationStageTests : StreamTestBase
 {
@@ -35,10 +35,8 @@ private async Task<List<HttpResponseMessage>> RunStageAsync(
             var resSrc = b.Add(responseSource);
             var signalSink = b.Add(Sink.Ignore<IControlItem>().MapMaterializedValue(_ => NotUsed.Instance));
 
-            var resetSrc = b.Add(Source.Never<NotUsed>());
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(resetSrc).To(corr.InReset);
             b.From(corr.OutResponse).To(s);
             b.From(corr.OutControl).To(signalSink);
 
@@ -108,17 +106,17 @@ public async Task Should_PreserveRequestReference_WhenCorrelated()
             "response.RequestMessage must be the exact same object reference as the sent request.");
     }
 
-    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3-11CR-004: Response arrives before request → correctly buffered and correlated")]
-    public async Task Should_BufferAndCorrelate_WhenResponseArrivesBeforeRequest()
+    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3-11CR-004: Request arrives, response arrives later → correctly correlated")]
+    public async Task Should_Correlate_WhenResponseArrivesAfterRequest()
     {
         var request = new HttpRequestMessage(HttpMethod.Get, "http://example.com/delayed");
         var response = OkResponse();
 
-        // Response source emits immediately; request source delayed by 300ms.
-        // The response will be buffered in the _waiting queue until the request arrives.
+        // Request is sent first; response is delayed by 300ms.
+        // Stage pulls InRequest first, then pulls InResponse after request arrives.
         var results = await RunStageAsync(
-            Source.Single(request).InitialDelay(TimeSpan.FromMilliseconds(300)),
-            Source.Single(response));
+            Source.Single(request),
+            Source.Single(response).InitialDelay(TimeSpan.FromMilliseconds(300)));
 
         Assert.Single(results);
         Assert.Same(request, results[0].RequestMessage);
@@ -132,7 +130,7 @@ public async Task Should_BufferAndCorrelate_WhenRequestArrivesBeforeResponse()
         var response = OkResponse();
 
         // Request source emits immediately; response source delayed by 300ms.
-        // The request will be buffered in the _pending queue until the response arrives.
+        // Stage waits for the response after pulling and storing the request.
         var results = await RunStageAsync(
             Source.Single(request),
             Source.Single(response).InitialDelay(TimeSpan.FromMilliseconds(300)));
@@ -164,10 +162,8 @@ public async Task Should_StayAlive_WhenQueuesEmptyButUpstreamOpen()
             var resSrc = b.Add(responseSource);
             var signalSink = b.Add(Sink.Ignore<IControlItem>().MapMaterializedValue(_ => NotUsed.Instance));
 
-            var resetSrc = b.Add(Source.Never<NotUsed>());
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(resetSrc).To(corr.InReset);
             b.From(corr.OutResponse).To(s);
             b.From(corr.OutControl).To(signalSink);
 
@@ -180,19 +176,19 @@ public async Task Should_StayAlive_WhenQueuesEmptyButUpstreamOpen()
         await Assert.ThrowsAsync<TimeoutException>(() => task.WaitAsync(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken));
     }
 
-    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3-11CR-007: Stage remains open while pending requests still exist")]
-    public async Task Should_RemainOpen_WhenPendingRequestsExist()
+    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3-11CR-007: Stage remains open while in-flight request awaits response")]
+    public async Task Should_RemainOpen_WhenInFlightRequestAwaitingResponse()
     {
-        // Send 2 requests but only 1 response.
-        // The response source stays open (via Concat+Never) so the stage cannot
-        // complete due to upstream finish — only the pending request queue keeps it alive.
+        // Send 2 requests but keep the response source open after delivering only 1 response.
+        // The second request is pulled after the first response; it waits for its response
+        // indefinitely (Never) — the stage must remain open.
         var request1 = new HttpRequestMessage(HttpMethod.Get, "http://example.com/1");
         var request2 = new HttpRequestMessage(HttpMethod.Get, "http://example.com/2");
         var response1 = OkResponse();
 
         var sink = Sink.Seq<HttpResponseMessage>();
 
-        // Keep the response source open so only pending-request logic matters.
+        // Keep the response source open so only the in-flight back-pressure matters.
         var neverEndingResponses = Source.Single(response1)
             .Concat(Source.Never<HttpResponseMessage>());
 
@@ -203,10 +199,8 @@ public async Task Should_RemainOpen_WhenPendingRequestsExist()
             var resSrc = b.Add(neverEndingResponses);
             var signalSink = b.Add(Sink.Ignore<IControlItem>().MapMaterializedValue(_ => NotUsed.Instance));
 
-            var resetSrc = b.Add(Source.Never<NotUsed>());
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(resetSrc).To(corr.InReset);
             b.From(corr.OutResponse).To(s);
             b.From(corr.OutControl).To(signalSink);
 
@@ -215,8 +209,7 @@ public async Task Should_RemainOpen_WhenPendingRequestsExist()
 
         var task = graph.Run(Materializer);
 
-        // The stream should NOT complete because there is still a pending request
-        // with no matching response. Wait briefly and verify it's still running.
+        // The stream should NOT complete because request2 is in flight with no matching response.
         var completed = task.WaitAsync(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken);
 
         await Assert.ThrowsAsync<TimeoutException>(() => completed);
@@ -237,10 +230,8 @@ public async Task Should_EmitOneStreamAcquireItem_WhenSingleRequestPushed()
             var resSrc = b.Add(Source.Single(response));
             var responseSink = b.Add(Sink.Ignore<HttpResponseMessage>().MapMaterializedValue(_ => NotUsed.Instance));
 
-            var resetSrc = b.Add(Source.Never<NotUsed>());
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(resetSrc).To(corr.InReset);
             b.From(corr.OutResponse).To(responseSink);
             b.From(corr.OutControl).To(s);
 
@@ -278,10 +269,8 @@ public async Task Should_EmitTwoStreamAcquireItems_WhenTwoRequestsPushed()
             var resSrc = b.Add(Source.From(responses));
             var responseSink = b.Add(Sink.Ignore<HttpResponseMessage>().MapMaterializedValue(_ => NotUsed.Instance));
 
-            var resetSrc = b.Add(Source.Never<NotUsed>());
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(resetSrc).To(corr.InReset);
             b.From(corr.OutResponse).To(responseSink);
             b.From(corr.OutControl).To(s);
 
diff --git a/src/TurboHttp.StreamTests/RFC9112/14_Http11PipelineReconnectTests.cs b/src/TurboHttp.StreamTests/RFC9112/14_Http11PipelineReconnectTests.cs
@@ -8,29 +8,22 @@
 namespace TurboHttp.StreamTests.RFC9112;
 
 /// <summary>
-/// Tests the no-pipelining-after-reconnect guard per RFC 9112 §9.3.2.
-/// Verifies that the first request after (re)connect waits for a response before pipelining is allowed,
-/// and that reconnect resets the guard.
+/// Tests the strict one-request-in-flight back-pressure contract per RFC 9112 §9.
+/// Verifies that each request waits for its own response before the next request is pulled.
 /// </summary>
 /// <remarks>
 /// Stage under test: <see cref="Http1XCorrelationStage"/>.
-/// RFC 9112 §9.3.2: A client SHOULD NOT pipeline requests on a newly opened connection
-/// until it knows the connection is persistent.
+/// RFC 9112 §9: HTTP/1.x requests and responses MUST be sent and received in order.
+/// The InReset inlet has been removed; strict serial back-pressure is always enforced.
 /// </remarks>
 public sealed class Http11PipelineReconnectTests : StreamTestBase
 {
     private static HttpResponseMessage OkResponse()
         => new(HttpStatusCode.OK);
 
-    /// <summary>
-    /// Builds and runs a closed graph that wires requestSource → InRequest, responseSource → InResponse,
-    /// resetSource → InReset, Out → responseSink, OutSignal → signalSink.
-    /// Returns the collected signals and responses once the stream completes.
-    /// </summary>
     private async Task<(List<IControlItem> Signals, List<HttpResponseMessage> Responses)> RunStageAsync(
         Source<HttpRequestMessage, NotUsed> requestSource,
         Source<HttpResponseMessage, NotUsed> responseSource,
-        Source<NotUsed, NotUsed> resetSource,
         TimeSpan? timeout = null)
     {
         var signalSink = Sink.Seq<IControlItem>();
@@ -41,11 +34,9 @@ private static HttpResponseMessage OkResponse()
             var corr = b.Add(new Http1XCorrelationStage());
             var reqSrc = b.Add(requestSource);
             var resSrc = b.Add(responseSource);
-            var rstSrc = b.Add(resetSource);
 
             b.From(reqSrc).To(corr.InRequest);
             b.From(resSrc).To(corr.InResponse);
-            b.From(rstSrc).To(corr.InReset);
             b.From(corr.OutControl).To(sigSink);
             b.From(corr.OutResponse).To(resSink);
 
@@ -59,113 +50,31 @@ private static HttpResponseMessage OkResponse()
         return (signals.ToList(), responses.ToList());
     }
 
-    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3.2-PL-001: First request after connect is not pipelined")]
-    public async Task Should_WaitForResponse_When_FirstRequestAfterConnect()
+    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3.2-PL-001: Each request waits for its response — two requests yield two signals")]
+    public async Task Should_EmitSignalPerRequest_WhenEachRequestWaitsForResponse()
     {
         // Two requests emitted immediately, responses delayed and spaced.
-        // Guard locks after req1 → req2 cannot be pulled until resp1 arrives and unlocks the guard.
-        // After unlock, req2 becomes first-pending → signal #2.
+        // With strict back-pressure: req1 is pulled → blocked on InResponse.
+        // resp1 arrives → req1 paired, req2 pulled → signal #2.
+        // resp2 arrives → req2 paired.
         var requests = new[]
         {
             new HttpRequestMessage(HttpMethod.Get, "http://example.com/1"),
             new HttpRequestMessage(HttpMethod.Get, "http://example.com/2")
         };
 
-        // resp1 at ~200ms, resp2 at ~800ms — spaced so req2 processes before resp2 arrives
+        // resp1 at ~200ms, resp2 at ~800ms
         var responseSource = Source.Single(OkResponse())
             .InitialDelay(TimeSpan.FromMilliseconds(200))
             .Concat(Source.Single(OkResponse())
                 .InitialDelay(TimeSpan.FromMilliseconds(600)));
 
         var (signals, responses) = await RunStageAsync(
             Source.From(requests),
-            responseSource,
-            Source.Never<NotUsed>());
+            responseSource);
 
         Assert.Equal(2, responses.Count);
         Assert.Equal(2, signals.Count);
         Assert.All(signals, s => Assert.IsType<StreamAcquireItem>(s));
     }
-
-    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3.2-PL-002: Second request may pipeline after first response")]
-    public async Task Should_Pipeline_When_FirstResponseReceived()
-    {
-        // Three requests emitted immediately, responses delayed and spaced.
-        // Guard locks after req1 → req2 blocked.
-        // At ~200ms: resp1 → unlock → pull req2 → req2 arrives (signal #2), pull req3 (unlocked).
-        // req3 arrives with pending=1 → NO signal (pipelined behind req2).
-        // At ~800ms: resp2+resp3 → correlate remaining.
-        // Result: 2 signals proves pipelining occurred (req3 did not become first-pending).
-        var requests = new[]
-        {
-            new HttpRequestMessage(HttpMethod.Get, "http://example.com/1"),
-            new HttpRequestMessage(HttpMethod.Get, "http://example.com/2"),
-            new HttpRequestMessage(HttpMethod.Get, "http://example.com/3")
-        };
-
-        // resp1 at ~200ms, resp2+resp3 at ~800ms — req2+req3 arrive and pipeline before resp2
-        var responseSource = Source.Single(OkResponse())
-            .InitialDelay(TimeSpan.FromMilliseconds(200))
-            .Concat(Source.From(new[] { OkResponse(), OkResponse() })
-                .InitialDelay(TimeSpan.FromMilliseconds(600)));
-
-        var (signals, responses) = await RunStageAsync(
-            Source.From(requests),
-            responseSource,
-            Source.Never<NotUsed>());
-
-        Assert.Equal(3, responses.Count);
-        // req1 → signal (first-pending, guard locked)
-        // req2 → signal (first-pending after resp1 unlocks guard)
-        // req3 → no signal (pipelined behind req2 while guard is unlocked)
-        Assert.Equal(2, signals.Count);
-        Assert.All(signals, s => Assert.IsType<StreamAcquireItem>(s));
-    }
-
-    [Fact(Timeout = 10_000, DisplayName = "RFC9112-9.3.2-PL-003: Reconnect resets pipeline guard")]
-    public async Task Should_ResetGuard_When_ConnectionReestablished()
-    {
-        // req1 immediate, req2+req3 delayed 400ms (arrive after reset).
-        // resp1 at ~100ms, resp2+resp3 at ~600ms.
-        // Reset at ~250ms (after resp1 unlocks, before req2 arrives).
-        //
-        // Timeline:
-        //   t=0:   req1 → signal #1, guard locked
-        //   t=100: resp1 → correlate → unlock → pull (no req available yet)
-        //   t=250: reset → re-lock guard
-        //   t=400: req2 → signal #2, guard locked → don't pull req3
-        //   t=600: resp2 → correlate → unlock → pull req3
-        //          req3 → signal #3
-        //          resp3 → correlate
-        //
-        // Result: 3 signals proves reset re-locked the guard.
-        var req1 = new HttpRequestMessage(HttpMethod.Get, "http://example.com/1");
-        var req2 = new HttpRequestMessage(HttpMethod.Get, "http://example.com/2");
-        var req3 = new HttpRequestMessage(HttpMethod.Get, "http://example.com/3");
-
-        var requestSource = Source.Single(req1)
-            .Concat(Source.From(new[] { req2, req3 })
-                .InitialDelay(TimeSpan.FromMilliseconds(400)));
-
-        var responseSource = Source.Single(OkResponse())
-            .InitialDelay(TimeSpan.FromMilliseconds(100))
-            .Concat(Source.From(new[] { OkResponse(), OkResponse() })
-                .InitialDelay(TimeSpan.FromMilliseconds(500)));
-
-        var resetSource = Source.Single(NotUsed.Instance)
-            .InitialDelay(TimeSpan.FromMilliseconds(250));
-
-        var (signals, responses) = await RunStageAsync(
-            requestSource,
-            responseSource,
-            resetSource);
-
-        Assert.Equal(3, responses.Count);
-        // req1 → signal (first-pending, guard locked)
-        // reset → guard re-locked
-        // req2 → signal (first-pending after reset)
-        // req3 → signal (first-pending after resp2 unlocks, guard was re-locked)
-        Assert.Equal(3, signals.Count);
-        Assert.All(signals, s => Assert.IsType<StreamAcquireItem>(s));
-    }
 }
diff --git a/src/TurboHttp/Streams/Http10Engine.cs b/src/TurboHttp/Streams/Http10Engine.cs
@@ -28,9 +28,6 @@ public BidiFlow<HttpRequestMessage, IOutputItem, IInputItem, HttpResponseMessage
 
             b.From(decoder.Outlet).To(correlation.InResponse);
 
-            var resetSrc = b.Add(Source.Empty<NotUsed>());
-            b.From(resetSrc).To(correlation.InReset);
-
             return new BidiShape<
                 HttpRequestMessage,
                 IOutputItem,
diff --git a/src/TurboHttp/Streams/Http11Engine.cs b/src/TurboHttp/Streams/Http11Engine.cs
@@ -30,9 +30,6 @@ public BidiFlow<HttpRequestMessage, IOutputItem, IInputItem, HttpResponseMessage
 
             b.From(decoder.Outlet).To(correlation.InResponse);
 
-            var resetSrc = b.Add(Source.Empty<NotUsed>());
-            b.From(resetSrc).To(correlation.InReset);
-
             var signalCast = b.Add(Flow.Create<IControlItem>().Select(IOutputItem (x) => x));
 
             var batchFlow = b.Add(
diff --git a/src/TurboHttp/Streams/Stages/Routing/Http1XCorrelationStage.cs b/src/TurboHttp/Streams/Stages/Routing/Http1XCorrelationStage.cs
