refactor: complexity cleanup across protocol state machines, body pumps, and stages (+3 bug fixes)

src/GaudiHTTP.Tests/Pooling/ConnectionObjectPoolSpec.cs

@@ -61,29 +61,29 @@ public void Dispose_on_a_Poolable_returns_it_to_the_singleton_pool_reset()
         ctx.Return(b);
     }
 
-    private sealed class NeverRentedCounter : Poolable<NeverRentedCounter>
+    private sealed class ReturnFirstCounter : Poolable<ReturnFirstCounter>
     {
         public int Value;
-        public int ResetCount;
-        protected override void OnReset()
-        {
-            Value = 0;
-            ResetCount++;
-        }
+        protected override void OnReset() => Value = 0;
     }
 
     [Fact(Timeout = 5000)]
-    public void Return_without_prior_rent_resets_and_discards_instead_of_throwing()
+    public void Return_before_any_rent_must_not_throw_and_later_rent_reuses_the_instance()
     {
-        // Directly-constructed poolables (tests injecting fakes, abort paths racing the first
-        // Rent<T> of the process) dispose without a pool existing for their type. That must
-        // behave like returning to a full pool — reset and drop — not throw.
-        var a = new NeverRentedCounter { Value = 42 };
+        // A pooled object constructed directly (new, not rented) disposes itself back into the
+        // pool. That Return may be the very first pool touch for the type — it must not require
+        // a factory registration.
+        var ctx = ConnectionObjectPool.Instance;
+        var a = new ReturnFirstCounter { Value = 5 };
 
         a.Dispose();
 
-        Assert.Equal(1, a.ResetCount);
-        Assert.Equal(0, a.Value);
+        var b = ctx.Rent(static () => new ReturnFirstCounter());
+
+        Assert.Same(a, b);
+        Assert.Equal(0, b.Value);
+
+        ctx.Return(b);
     }
 
     [Fact(Timeout = 5000)]

src/GaudiHTTP.Tests/Protocol/Body/FlowControlledBodyPumpSpec.cs

@@ -1,3 +1,5 @@
+using System.Collections;
+using System.Reflection;
 using Akka.Actor;
 using GaudiHTTP.Protocol.Body;
 using GaudiHTTP.Protocol.Syntax.Http2;
@@ -89,9 +91,6 @@ private static void DrainToCompletion(FlowControlledBodyPump scheduler, FakeTarg
                 case BodyReadComplete<int> rc:
                     scheduler.HandleReadComplete(rc.StreamId, rc.BytesRead);
                     break;
-                case BodyReadContinue<int> dc:
-                    scheduler.HandleBodyReadContinue(dc.StreamId);
-                    break;
             }
         }
     }
@@ -369,6 +368,77 @@ public void OrphanRefund_should_return_reserved_window_when_stream_cancelled_dur
         Assert.Equal(windowBefore, flow.ConnectionSendWindow);
     }
 
+    [Fact(Timeout = 5000)]
+    public void FailedRead_should_refund_reserved_window()
+    {
+        var target = new FakeTarget();
+        var flow = MakeFlow();
+        var scheduler = new FlowControlledBodyPump(target, flow, new CancellationTokenSource(), chunkSize: 1 * 1024, hardCap: 16);
+
+        flow.InitStreamSendWindow(1);
+        var windowBefore = flow.ConnectionSendWindow;
+
+        // Async read path so the reservation is held while the read is in flight.
+        var tcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var blockingStream = new DelegatingReadStream((_, _) => new ValueTask<int>(tcs.Task));
+        scheduler.Register(1, blockingStream, null, CancellationToken.None);
+
+        Assert.Equal(windowBefore - 1 * 1024, flow.ConnectionSendWindow);
+
+        // Read fails while the stream is still active (not orphaned): nothing was
+        // sent, so the full reservation must be refunded.
+        scheduler.HandleReadFailed(1, new IOException("read failed"));
+
+        Assert.Single(target.Failed);
+        Assert.Equal(windowBefore, flow.ConnectionSendWindow);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void Cancel_should_dispose_queued_non_inflight_slot_immediately_even_when_window_stays_exhausted()
+    {
+        var target = new FakeTarget();
+        var flow = MakeFlow(connWindow: 65535);
+        // hardCap = 1: only one async read may be in flight at a time.
+        var scheduler = new FlowControlledBodyPump(target, flow, new CancellationTokenSource(), chunkSize: 64, hardCap: 1);
+
+        flow.InitStreamSendWindow(1);
+        flow.InitStreamSendWindow(3);
+
+        // Stream 1 occupies the only read slot with a read that never completes.
+        var tcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var blockingStream = new DelegatingReadStream((_, _) => new ValueTask<int>(tcs.Task));
+        scheduler.Register(1, blockingStream, null, CancellationToken.None);
+
+        // Stream 3 has an open window, so it is enqueued — but _asyncInFlight (1) already meets
+        // hardCap (1), so it never starts a read: it sits in the ready queue, not in-flight and
+        // not window-blocked.
+        scheduler.Register(3, MakeBody(100), 100, CancellationToken.None);
+
+        var activeSlots = GetActiveSlots(scheduler);
+        Assert.True(activeSlots.Contains(3));
+
+        // Cancel stream 3 while queued. The connection window never opens again afterward,
+        // simulating a stalled connection — a deferred-dispose discipline would leave the
+        // cancelled slot (and its pooled buffer) alive until Cleanup(); the correct discipline
+        // disposes it immediately, matching MultiplexedBodyPump.
+        scheduler.Cancel(3);
+
+        Assert.False(activeSlots.Contains(3));
+
+        // Opening the window afterward must not resurrect or emit anything for stream 3.
+        flow.OnSendWindowUpdate(0, 65535);
+        scheduler.OnWindowUpdate(0);
+
+        Assert.DoesNotContain(3, target.Completed);
+        Assert.DoesNotContain(target.Emitted, e => e.StreamId == 3);
+    }
+
+    private static IDictionary GetActiveSlots(FlowControlledBodyPump scheduler)
+    {
+        var field = typeof(FlowControlledBodyPump).GetField("_activeSlots", BindingFlags.NonPublic | BindingFlags.Instance)!;
+        return (IDictionary)field.GetValue(scheduler)!;
+    }
+
     [Fact(Timeout = 5000)]
     public void WindowBlocked_should_block_when_available_below_half_chunk()
     {

src/GaudiHTTP.Tests/Protocol/Body/MultiplexedBodyPumpSpec.cs

@@ -64,9 +64,6 @@ private static void DrainToCompletion(
                 case BodyReadComplete<long> rc:
                     pump.HandleReadComplete(rc.StreamId, rc.BytesRead);
                     break;
-                case BodyReadContinue<long> dc:
-                    pump.HandleBodyReadContinue(dc.StreamId);
-                    break;
             }
         }
     }
@@ -91,9 +88,6 @@ private static void DriveReadsWithoutCapacity(
                 case BodyReadComplete<long> rc:
                     pump.HandleReadComplete(rc.StreamId, rc.BytesRead);
                     break;
-                case BodyReadContinue<long> dc:
-                    pump.HandleBodyReadContinue(dc.StreamId);
-                    break;
             }
         }
     }
@@ -199,19 +193,18 @@ public void SyncFastPath_should_drain_small_body_inline()
     }
 
     [Fact(Timeout = 5000)]
-    public void Sync_reads_should_drain_body_larger_than_starvation_threshold()
+    public void Many_consecutive_sync_reads_should_drain_fully()
     {
         var target = new FakeTarget();
-        // Use small chunk so many sync reads happen
+        // Small chunk so the body needs dozens of consecutive sync reads to drain.
         var pump = MakePump(target, chunkSize: 16);
 
         pump.Register(1L, MakeBody(65 * 16), contentLength: null, CancellationToken.None);
-        // Starvation guard fires at 64 consecutive reads and dispatches BodyReadContinue.
-        // DrainToCompletion processes those messages to drive the pump to completion.
         DrainToCompletion(pump, target);
 
         var total = target.Emitted.Where(e => !e.EndStream).Sum(e => e.Data.Length);
-        Assert.True(total > 0, "Expected at least some data emitted before starvation guard fired");
+        Assert.Equal(65 * 16, total);
+        Assert.Single(target.Completed);
     }
 
     [Fact(Timeout = 5000)]

src/GaudiHTTP.Tests/Protocol/Body/SerialBodyPumpSpec.cs

@@ -60,17 +60,14 @@ private static void DrainToCompletion(
                 case BodyReadComplete<int> rc:
                     pump.HandleReadComplete(rc.BytesRead);
                     break;
-                case BodyReadContinue<int>:
-                    pump.HandleBodyReadContinue();
-                    break;
             }
         }
     }
 
     /// <summary>
     /// Target that calls OnCapacityAvailable() synchronously after EmitDataFrames,
     /// simulating a consumer that immediately signals capacity after each chunk.
-    /// Uses a direct-dispatch actor ref so pump messages (BodyReadComplete, BodyReadContinue)
+    /// Uses a direct-dispatch actor ref so pump messages (BodyReadComplete)
     /// are processed inline as they arrive — no external drain loop needed.
     /// </summary>
     private sealed class AutoResumeTarget : IBodyDrainTarget
@@ -119,9 +116,6 @@ protected override void TellInternal(object message, IActorRef sender)
                     case BodyReadComplete<int> rc:
                         owner._pump.HandleReadComplete(rc.BytesRead);
                         break;
-                    case BodyReadContinue<int>:
-                        owner._pump.HandleBodyReadContinue();
-                        break;
                 }
             }
         }
@@ -193,9 +187,6 @@ public void OnCapacityAvailable_should_resume_drain_after_capacity_exhaustion()
                 case BodyReadComplete<int> rc:
                     pump.HandleReadComplete(rc.BytesRead);
                     break;
-                case BodyReadContinue<int>:
-                    pump.HandleBodyReadContinue();
-                    break;
             }
         }
 
@@ -217,9 +208,6 @@ public void OnCapacityAvailable_should_resume_drain_after_capacity_exhaustion()
                     case BodyReadComplete<int> rc:
                         pump.HandleReadComplete(rc.BytesRead);
                         break;
-                    case BodyReadContinue<int>:
-                        pump.HandleBodyReadContinue();
-                        break;
                 }
             }
         }
@@ -299,19 +287,17 @@ public void Large_body_should_drain_fully_with_auto_resume()
     }
 
     [Fact(Timeout = 5000)]
-    public void Sync_reads_should_complete_large_body_with_starvation_guard_via_actor()
+    public void Many_consecutive_sync_reads_should_drain_fully_with_auto_resume()
     {
         var target = new AutoResumeTarget();
-        // Small chunk to trigger many sync reads; starvation guard will fire at 64 consecutive reads.
-        // The AutoResumeTarget processes BodyReadContinue messages inline, so the drain completes fully.
+        // Small chunk so the body needs dozens of consecutive sync reads to drain.
         var pump = MakePump(target, chunkSize: 16, maxCapacity: 2);
         target.SetPump(pump);
         var totalSize = 65 * 16;
         var body = MakeBody(totalSize);
 
         pump.Register(body, contentLength: null, CancellationToken.None);
 
-        // AutoResumeTarget processes BodyReadContinue via DrainMessages() — full body drains.
         var emittedBytes = target.Emitted.Where(e => !e.EndStream).Sum(e => e.Data.Length);
         Assert.Equal(totalSize, emittedBytes);
         Assert.Single(target.Completed);
@@ -350,9 +336,6 @@ public void Capacity_grant_during_pending_completion_must_not_corrupt_buffer()
                 case BodyReadComplete<int> rc:
                     pump.HandleReadComplete(rc.BytesRead);
                     break;
-                case BodyReadContinue<int>:
-                    pump.HandleBodyReadContinue();
-                    break;
             }
         }