feat(statemachine): Http10ServerStateMachine defers encode until body bytes delivered

Replace the old ResponseBodyHandle/timer-based body drain with the same
two-phase deferred encode pattern used in Http10ClientStateMachine.
OnResponse calls Encode (always returns 0), stores the deferredResponse,
then OnBodyMessage dispatches OutboundBodyChunk/Complete/Failed to call
EncodeDeferred and emit TransportData once the full body is available.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: st0o0

src/TurboHTTP.Tests/Http10/Http10ServerStateMachineSpec.cs

@@ -1,4 +1,8 @@
+using System.Net;
+using System.Text;
+using Akka.Actor;
 using Servus.Akka.Transport;
+using TurboHTTP.Protocol;
 using TurboHTTP.Protocol.Syntax.Http10;
 using TurboHTTP.Protocol.Syntax.Http10.Server;
 using TurboHTTP.Tests.Shared;
@@ -50,17 +54,61 @@ public void DecodeClientData_should_mark_should_complete()
 
     [Fact(Timeout = 5000)]
     [Trait("RFC", "RFC1945")]
-    public void OnResponse_should_emit_encoded_response()
+    public void OnResponse_should_not_emit_transport_data_before_body_delivered()
     {
         var ops = MakeOps();
         var sm = new Http10ServerStateMachine(ops);
 
-        var response = new HttpResponseMessage(System.Net.HttpStatusCode.OK);
-        response.Content = new StringContent("test body");
+        var response = new HttpResponseMessage(HttpStatusCode.OK)
+        {
+            Content = new StringContent("test body")
+        };
 
         sm.OnResponse(response);
 
-        Assert.Single(ops.Outbound.OfType<TransportData>());
+        Assert.DoesNotContain(ops.Outbound, o => o is TransportData);
+    }
+
+    [Fact(Timeout = 5000)]
+    [Trait("RFC", "RFC1945")]
+    public async Task OnResponse_with_body_should_emit_transport_data_after_body_chunk()
+    {
+        var system = ActorSystem.Create("http10-server-sm-test");
+        try
+        {
+            var inbox = Inbox.Create(system);
+            var ops = new FakeServerOps { StageActor = inbox.Receiver };
+            var sm = new Http10ServerStateMachine(ops);
+            sm.PreStart();
+
+            var response = new HttpResponseMessage(HttpStatusCode.OK)
+            {
+                Content = new ByteArrayContent(Encoding.UTF8.GetBytes("hello"))
+            };
+            sm.OnResponse(response);
+
+            // No TransportData yet (deferred)
+            Assert.DoesNotContain(ops.Outbound, o => o is TransportData);
+
+            // Receive the OutboundBodyChunk and feed it back
+            var msg = await Task.Run(() => inbox.Receive(TimeSpan.FromSeconds(3)));
+            var chunk = Assert.IsType<OutboundBodyChunk>(msg);
+            sm.OnBodyMessage(chunk);
+
+            var msg2 = await Task.Run(() => inbox.Receive(TimeSpan.FromSeconds(3)));
+            sm.OnBodyMessage(msg2); // OutboundBodyComplete
+
+            // Now TransportData should exist
+            Assert.Contains(ops.Outbound, o => o is TransportData);
+            var td = ops.Outbound.OfType<TransportData>().First();
+            var text = Encoding.ASCII.GetString(td.Buffer.Memory.Span[..td.Buffer.Length]);
+            Assert.Contains("Content-Length: 5", text);
+            Assert.Contains("hello", text);
+        }
+        finally
+        {
+            system.Dispose();
+        }
     }
 
     [Fact(Timeout = 5000)]

src/TurboHTTP/Protocol/Syntax/Http10/Server/Http10ServerStateMachine.cs

@@ -1,20 +1,22 @@
+using System.Buffers;
 using Servus.Akka.Transport;
 using TurboHTTP.Protocol.Syntax.Http10.Options;
 using TurboHTTP.Streams;
+using static Servus.Core.Servus;
 
 namespace TurboHTTP.Protocol.Syntax.Http10.Server;
 
 internal sealed class Http10ServerStateMachine : IServerStateMachine
 {
-    private const string DrainBodyTimerKey = "drain-body";
-
     private readonly IServerStageOperations _ops;
     private readonly Http10ServerDecoder _decoder;
     private readonly Http10ServerEncoder _encoder;
     private readonly long _maxRequestBodySize;
 
     private RequestBodyHandle? _activeBody;
-    private ResponseBodyHandle? _responseBody;
+    private HttpResponseMessage? _deferredResponse;
+    private IMemoryOwner<byte>? _deferredBodyOwner;
+    private int _deferredBodyLength;
 
     public bool CanAcceptResponse => true;
     public bool ShouldComplete { get; private set; }
@@ -71,17 +73,30 @@ public void OnResponse(HttpResponseMessage response)
         response.Headers.Connection.Clear();
         response.Headers.Connection.Add("close");
 
-        var responseBuffer = TransportBuffer.Rent(8192);
-        var span = responseBuffer.FullMemory.Span;
-        var written = _encoder.Encode(span, response, _ops.StageActor);
-        responseBuffer.Length = written;
-        _ops.OnOutbound(new TransportData(responseBuffer));
-
-        _responseBody = new ResponseBodyHandle(
-            response.Content,
-            isChunked: false,
-            onDataReady: () => _ops.OnScheduleTimer(DrainBodyTimerKey, TimeSpan.Zero));
-        _responseBody.StartDrain();
+        // Http10ServerEncoder always returns 0 — it starts the buffered body encoder
+        // and sends OutboundBodyChunk/OutboundBodyComplete to the stage actor.
+        var tempBuffer = TransportBuffer.Rent(1);
+        try
+        {
+            var written = _encoder.Encode(tempBuffer.FullMemory.Span, response, _ops.StageActor);
+            if (written > 0)
+            {
+                // Synchronous path (not currently used, kept for safety)
+                tempBuffer.Length = written;
+                _ops.OnOutbound(new TransportData(tempBuffer));
+                return;
+            }
+        }
+        catch
+        {
+            tempBuffer.Dispose();
+            throw;
+        }
+
+        tempBuffer.Dispose();
+
+        // Deferred — waiting for OutboundBodyChunk + OutboundBodyComplete via OnBodyMessage
+        _deferredResponse = response;
     }
 
     public void OnDownstreamFinished()
@@ -90,63 +105,61 @@ public void OnDownstreamFinished()
 
     public void OnTimerFired(string name)
     {
-        if (name == DrainBodyTimerKey)
-        {
-            DrainResponseBody();
-        }
-    }
-
-    public void OnBodyMessage(object msg) { }
-
-    public void Cleanup()
-    {
-        _activeBody?.Abort(new OperationCanceledException("Connection closed"));
-        _activeBody = null;
-
-        _responseBody?.Abort();
-        _responseBody = null;
     }
 
-    private void DrainResponseBody()
+    public void OnBodyMessage(object msg)
     {
-        if (_responseBody is null)
+        switch (msg)
         {
-            return;
-        }
-
-        while (_responseBody.TryRead(out var result))
-        {
-            var data = result.Buffer;
-            if (data.Length > 0)
-            {
-                var outBuf = TransportBuffer.Rent((int)data.Length);
-                var span = outBuf.FullMemory.Span;
-
-                if (data.IsSingleSegment)
+            case OutboundBodyChunk chunk when _deferredResponse is not null:
+                _deferredBodyOwner?.Dispose();
+                _deferredBodyOwner = chunk.Owner;
+                _deferredBodyLength = chunk.Length;
+                break;
+
+            case OutboundBodyComplete when _deferredResponse is not null && _deferredBodyOwner is not null:
+                TransportBuffer? item = null;
+                try
                 {
-                    data.FirstSpan.CopyTo(span);
+                    var body = _deferredBodyOwner.Memory.Span[.._deferredBodyLength];
+                    var bufferSize = 8192 + _deferredBodyLength;
+                    item = TransportBuffer.Rent(bufferSize);
+                    var written = _encoder.EncodeDeferred(item.FullMemory.Span, _deferredResponse, body);
+                    item.Length = written;
+                    _ops.OnOutbound(new TransportData(item));
                 }
-                else
+                catch (Exception ex)
                 {
-                    var offset = 0;
-                    foreach (var segment in data)
-                    {
-                        segment.Span.CopyTo(span[offset..]);
-                        offset += segment.Length;
-                    }
+                    item?.Dispose();
+                    Tracing.For("Protocol").Error(this, "Failed to encode HTTP/1.0 response body: {0}", ex.Message);
                 }
+                finally
+                {
+                    _deferredBodyOwner.Dispose();
+                    _deferredBodyOwner = null;
+                    _deferredResponse = null;
+                }
+                break;
 
-                outBuf.Length = (int)data.Length;
-                _ops.OnOutbound(new TransportData(outBuf));
-            }
-
-            _responseBody.AdvanceReader(data.End);
+            case OutboundBodyFailed failed:
+                _deferredBodyOwner?.Dispose();
+                _deferredBodyOwner = null;
+                if (_deferredResponse is not null)
+                {
+                    Tracing.For("Protocol").Error(this, "Failed to read HTTP/1.0 response body: {0}", failed.Reason.Message);
+                    _deferredResponse = null;
+                }
+                break;
         }
+    }
 
-        if (_responseBody.IsComplete)
-        {
-            _responseBody.Dispose();
-            _responseBody = null;
-        }
+    public void Cleanup()
+    {
+        _activeBody?.Abort(new OperationCanceledException("Connection closed"));
+        _activeBody = null;
+
+        _deferredBodyOwner?.Dispose();
+        _deferredBodyOwner = null;
+        _deferredResponse = null;
     }
 }