feat(conn): streaming response bodies — RespBody::Stream + chunked TE (v0.3 chunk 1)

Design decisions (per handoff Q1 user answer + Q3 proposal):

- Producer shape is PULL: the handler returns a smarm Receiver<Vec<u8>>
  (RespBody::Stream / StreamBody, From<Receiver<Vec<u8>>> for ergonomics);
  the producer is an actor the handler spawned. The conn actor pumps in
  pump_stream(): it keeps sole ownership of the socket and of write
  deadlines, and the recv() park between chunks is stoppable by the
  draining registry's request_stop (stop sentinel unwinds out of
  park_current; fd + registry guards clean up) — so an infinite stream is
  force-stoppable at the drain deadline like any in-flight request, with
  no polling. End of stream = every Sender dropped = terminating 0-chunk.
  Producer contract: a send() error means the conn died; exit.

- Framing: HTTP/1.1 gets transfer-encoding: chunked and the connection
  stays reusable after the terminator (keep-alive after chunked). HTTP/1.0
  has no chunked TE: bytes go raw, keep-alive is forced off, EOF delimits.
  User-set content-length/transfer-encoding headers are DROPPED for stream
  bodies — we own the framing, and CL+chunked is a smuggling vector.
  Empty producer chunks are skipped (a 0-length chunk would terminate the
  framing early).

- request_timeout stays READ-phase only (unchanged). NEW Config/ConnLimits
  field write_timeout (default 30s) gives every response write a per-write
  budget: the fixed head+body write, each streamed chunk, and the error
  paths (413/100-continue/4xx) all go through the now deadline-bounded
  write_all (wait_writable_timeout, mirroring read_some). Each chunk gets
  a FRESH budget — streams may outlive any whole-response clock; a single
  stalled write may not (write-side slowloris). Naming/default were
  flagged as a user call in the handoff: veto here if write_timeout(30s)
  isn't it.

- RespBody loses derive(Clone) (Receiver isn't Clone; Clone was unused)
  and gets a manual Debug.

Tests: 3 serialiser unit tests (chunked head shape, user-framing-header
stripping, 1.0 fallback) + 5 integration (chunked round-trip with decoder,
keep-alive after chunked, 1.0 EOF-delimited, shutdown force-stops an
infinite stream at drain deadline, stalled reader killed at write_timeout
with producer observing the closed channel).

src/conn.rs

@@ -146,21 +146,61 @@ impl Body {
 // RespBody — what plugs put on the wire.
 // ---------------------------------------------------------------------------
 //
-// Enum, not a trait, so the connection actor can pattern-match. Leaves room
-// for future variants like Sse, Stream, Chunked without touching the plug
-// API.
+// Enum, not a trait, so the connection actor can pattern-match.
+//
+// `Stream` is the pull-shaped streaming body (v0.3 design decision): the
+// handler hands back a `smarm::Receiver<Vec<u8>>` — typically the read end
+// of a channel whose `Sender` lives in a producer actor the handler
+// spawned — and the connection actor pumps it onto the wire. The conn
+// actor keeps sole ownership of the socket and of write deadlines; the
+// producer never touches the fd. End-of-stream is signalled by dropping
+// every `Sender` (the conn actor then emits the terminating 0-chunk).
+// Empty `Vec`s are skipped by the pump (a zero-length chunk would
+// terminate chunked framing early), so they are safe to send but useless.
 
-#[derive(Debug, Clone)]
 pub enum RespBody {
     Empty,
     Bytes(Vec<u8>),
+    Stream(StreamBody),
+}
+
+/// A streaming response body. Construct with [`StreamBody::new`] (or
+/// `RespBody::from(rx)`) and hand it to [`Conn::put_body`].
+///
+/// On HTTP/1.1 the connection actor sends it with
+/// `Transfer-Encoding: chunked` and the connection stays reusable after
+/// the stream completes. On HTTP/1.0 (no chunked framing) the bytes are
+/// written raw and the connection closes at end of stream to delimit the
+/// body.
+pub struct StreamBody {
+    pub rx: smarm::Receiver<Vec<u8>>,
+}
+
+impl StreamBody {
+    pub fn new(rx: smarm::Receiver<Vec<u8>>) -> Self {
+        Self { rx }
+    }
+}
+
+impl std::fmt::Debug for RespBody {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RespBody::Empty     => f.write_str("Empty"),
+            RespBody::Bytes(b)  => f.debug_tuple("Bytes").field(&b.len()).finish(),
+            RespBody::Stream(_) => f.write_str("Stream(..)"),
+        }
+    }
 }
 
 impl RespBody {
+    /// Body length for fixed bodies; 0 for `Stream` (a stream has no
+    /// length up front — that is the point — and is never serialised with
+    /// a `content-length`).
     pub fn len_hint(&self) -> usize {
         match self {
-            RespBody::Empty    => 0,
-            RespBody::Bytes(b) => b.len(),
+            RespBody::Empty     => 0,
+            RespBody::Bytes(b)  => b.len(),
+            RespBody::Stream(_) => 0,
         }
     }
 }
@@ -326,3 +366,11 @@ impl From<&'static str> for RespBody {
 impl From<&[u8]> for RespBody {
     fn from(s: &[u8]) -> Self { RespBody::Bytes(s.to_vec()) }
 }
+impl From<StreamBody> for RespBody {
+    fn from(s: StreamBody) -> Self { RespBody::Stream(s) }
+}
+impl From<smarm::Receiver<Vec<u8>>> for RespBody {
+    fn from(rx: smarm::Receiver<Vec<u8>>) -> Self {
+        RespBody::Stream(StreamBody::new(rx))
+    }
+}

src/conn_actor.rs

@@ -13,7 +13,7 @@
 //! `wait_readable` between bytes and `wait_writable` during slow writes;
 //! during those parks, other connection actors progress freely.
 
-use crate::conn::{Body, Conn, RespBody};
+use crate::conn::{Body, Conn, HttpVersion, RespBody, StreamBody};
 use crate::conn_registry::{Cast, ConnRegistry, DeregisterGuard};
 use crate::net::OwnedFd;
 use crate::parser::{self, ParseError};
@@ -50,7 +50,15 @@ pub struct ConnLimits {
     /// request until the request (head + body) is fully read. Expiry
     /// mid-head gets a best-effort 408; expiry mid-body just closes.
     /// Pipeline run time is NOT covered — that's the handler's business.
+    /// Covers the READ phase only; the write phase has its own
+    /// per-write budget (`write_timeout`) so a streaming response can
+    /// legitimately outlive any whole-request clock.
     pub request_timeout:    Duration,
+    /// Per-write budget for response bytes: every `write_all` (the fixed
+    /// head+body, and each streamed chunk) must complete within this.
+    /// A client that stops reading mid-response is dropped when its
+    /// socket buffer fills and a write stalls past the budget.
+    pub write_timeout:      Duration,
 }
 
 impl Default for ConnLimits {
@@ -62,6 +70,7 @@ impl Default for ConnLimits {
             max_body_bytes:     16 * 1024 * 1024,
             keep_alive_timeout: Duration::from_secs(60),
             request_timeout:    Duration::from_secs(30),
+            write_timeout:      Duration::from_secs(30),
         }
     }
 }
@@ -117,7 +126,7 @@ pub fn run_connection(
                 return;
             }
             Err(ReadHeadErr::Parse(e)) => {
-                emit_error_response(raw, &e);
+                emit_error_response(raw, &e, Instant::now() + limits.write_timeout);
                 return;
             }
         };
@@ -126,7 +135,11 @@ pub fn run_connection(
         // ----- 2. Read body. -----
         let body_len = parsed.content_length.unwrap_or(0);
         if body_len > limits.max_body_bytes {
-            let _ = write_all(raw, b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n");
+            let _ = write_all(
+                raw,
+                b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
+                Instant::now() + limits.write_timeout,
+            );
             return;
         }
 
@@ -134,7 +147,7 @@ pub fn run_connection(
         // body. RFC 7231 §5.1.1. We don't gate on app logic here; v1 always
         // accepts.
         if parsed.expect_100 {
-            if write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n").is_err() {
+            if write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n", Instant::now() + limits.write_timeout).is_err() {
                 return;
             }
         }
@@ -187,11 +200,29 @@ pub fn run_connection(
         }
 
         // ----- 4. Write the response. -----
-        let bytes = parser::serialise_response(&response_conn, keep_alive);
-        if write_all(raw, &bytes).is_err() {
+        // A Stream body on HTTP/1.0 has no chunked framing: the body is
+        // delimited by EOF, so keep-alive is forced off for this response
+        // (and `connection: close` goes on the wire).
+        let is_stream = matches!(response_conn.resp_body, RespBody::Stream(_));
+        let keep_alive = keep_alive
+            && !(is_stream && version == HttpVersion::Http10);
+
+        let head_bytes = parser::serialise_response(&response_conn, keep_alive);
+        if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
             return;
         }
 
+        if let RespBody::Stream(stream) = response_conn.resp_body {
+            let chunked = version == HttpVersion::Http11;
+            if pump_stream(raw, stream, chunked, limits.write_timeout).is_err() {
+                return;
+            }
+            if !chunked {
+                // EOF delimits the HTTP/1.0 stream body.
+                return;
+            }
+        }
+
         // ----- 5. Loop or close. -----
         if !keep_alive {
             return;
@@ -399,13 +430,24 @@ fn try_write_once(fd: RawFd, buf: &[u8]) {
 }
 
 // ---------------------------------------------------------------------------
-// write_all — robust write loop.
+// write_all — robust write loop, bounded by a deadline.
 // ---------------------------------------------------------------------------
+//
+// Mirrors read_some: each writability park is bounded by the remaining
+// budget. `ErrorKind::TimedOut` when the deadline passes before the bytes
+// are down — a client that stops reading must not pin this actor in
+// wait_writable forever (the write-side twin of slowloris).
 
-fn write_all(fd: RawFd, mut buf: &[u8]) -> io::Result<()> {
+fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
     while !buf.is_empty() {
-        // Park on writability before each syscall.
-        smarm::wait_writable(fd)?;
+        // Park on writability before each syscall, bounded by the budget.
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(io::Error::new(ErrorKind::TimedOut, "write deadline elapsed"));
+        }
+        if !smarm::wait_writable_timeout(fd, remaining)? {
+            return Err(io::Error::new(ErrorKind::TimedOut, "write deadline elapsed"));
+        }
 
         let n = unsafe {
             libc::write(fd, buf.as_ptr() as *const _, buf.len())
@@ -425,11 +467,68 @@ fn write_all(fd: RawFd, mut buf: &[u8]) -> io::Result<()> {
     Ok(())
 }
 
+// ---------------------------------------------------------------------------
+// pump_stream — drive a RespBody::Stream onto the wire.
+// ---------------------------------------------------------------------------
+//
+// The pull side of the v0.3 streaming design: the handler's producer actor
+// owns the Sender; this conn actor owns the socket and every write
+// deadline. We park in `recv()` between chunks — that park is stoppable
+// (a draining registry's `request_stop` unwinds us out of `park_current`
+// via the stop sentinel; fd + registry guards clean up), so an infinite
+// stream is force-stoppable at the drain deadline like any other in-flight
+// request. End of stream is the channel closing: every Sender dropped.
+//
+// Each chunk gets a FRESH write_timeout budget — a stream is expected to
+// outlive any whole-response clock; what is not tolerated is a single
+// write stalling. On write failure we return Err: the conn loop drops the
+// Receiver, and the producer's next `send` observes the closed channel and
+// should exit (that is the documented producer contract).
+//
+// `chunked` selects HTTP/1.1 chunked framing (hex-length CRLF payload
+// CRLF, terminated by a 0-chunk) vs HTTP/1.0 raw writes (EOF-delimited;
+// caller closes). Empty chunks are skipped — a zero-length chunk would
+// terminate the framing early.
+
+fn pump_stream(
+    fd: RawFd,
+    stream: StreamBody,
+    chunked: bool,
+    write_timeout: Duration,
+) -> io::Result<()> {
+    loop {
+        match stream.rx.recv() {
+            Ok(chunk) => {
+                if chunk.is_empty() {
+                    continue;
+                }
+                let deadline = Instant::now() + write_timeout;
+                if chunked {
+                    let mut framed = Vec::with_capacity(chunk.len() + 20);
+                    framed.extend_from_slice(format!("{:x}\r\n", chunk.len()).as_bytes());
+                    framed.extend_from_slice(&chunk);
+                    framed.extend_from_slice(b"\r\n");
+                    write_all(fd, &framed, deadline)?;
+                } else {
+                    write_all(fd, &chunk, deadline)?;
+                }
+            }
+            Err(_) => {
+                // All senders dropped: end of stream.
+                if chunked {
+                    write_all(fd, b"0\r\n\r\n", Instant::now() + write_timeout)?;
+                }
+                return Ok(());
+            }
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Error responses for unparseable / malformed requests.
 // ---------------------------------------------------------------------------
 
-fn emit_error_response(fd: RawFd, err: &ParseError) {
+fn emit_error_response(fd: RawFd, err: &ParseError, deadline: Instant) {
     let resp: &[u8] = match err {
         ParseError::TooManyHeaders =>
             b"HTTP/1.1 431 Request Header Fields Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
@@ -442,5 +541,5 @@ fn emit_error_response(fd: RawFd, err: &ParseError) {
         _ =>
             b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
     };
-    let _ = write_all(fd, resp);
+    let _ = write_all(fd, resp, deadline);
 }

src/lib.rs

@@ -28,7 +28,7 @@ pub mod conn_registry;
 pub mod serve;
 
 // Re-exports — what most users want at the crate root.
-pub use conn::{Assigns, Body, Conn, HeaderMap, HttpVersion, Method, Params, RespBody};
+pub use conn::{Assigns, Body, Conn, HeaderMap, HttpVersion, Method, Params, RespBody, StreamBody};
 pub use plug::{Next, Pipeline, Plug};
 pub use router::Router;
 pub use serve::{

src/parser.rs

@@ -187,8 +187,9 @@ pub fn build_conn(head: ParsedHead, body: Body) -> Conn {
 
 pub fn serialise_response(conn: &Conn, keep_alive: bool) -> Vec<u8> {
     // Pre-size: status line ~30 + headers ~50/each + body. Good enough.
-    let body_len = conn.resp_body.len_hint();
-    let mut out  = Vec::with_capacity(64 + conn.resp_headers.len() * 40 + body_len);
+    let body_len  = conn.resp_body.len_hint();
+    let is_stream = matches!(conn.resp_body, RespBody::Stream(_));
+    let mut out   = Vec::with_capacity(64 + conn.resp_headers.len() * 40 + body_len);
 
     let status = conn.status.unwrap_or(200);
     let reason = reason_phrase(status);
@@ -202,11 +203,17 @@ pub fn serialise_response(conn: &Conn, keep_alive: bool) -> Vec<u8> {
     out.extend_from_slice(b"\r\n");
 
     // User headers — written first so subsequent injection can skip them.
+    // For Stream bodies WE own the framing: a user `content-length` or
+    // `transfer-encoding` is dropped rather than emitted (the combination
+    // of content-length + chunked is a smuggling vector, and a stream has
+    // no length to promise anyway).
     let mut wrote_content_length = false;
     let mut wrote_connection     = false;
 
     for (name, value) in conn.resp_headers.iter() {
         match name {
+            "content-length" if is_stream => continue,
+            "transfer-encoding" if is_stream => continue,
             "content-length" => wrote_content_length = true,
             "connection"     => wrote_connection     = true,
             _ => {}
@@ -217,7 +224,15 @@ pub fn serialise_response(conn: &Conn, keep_alive: bool) -> Vec<u8> {
         out.extend_from_slice(b"\r\n");
     }
 
-    if !wrote_content_length {
+    if is_stream {
+        // HTTP/1.1: chunked framing, connection reusable afterwards.
+        // HTTP/1.0: no chunked TE exists; the body is raw bytes delimited
+        // by EOF — the caller passes keep_alive = false and we emit
+        // `connection: close` below.
+        if conn.version == HttpVersion::Http11 {
+            out.extend_from_slice(b"transfer-encoding: chunked\r\n");
+        }
+    } else if !wrote_content_length {
         out.extend_from_slice(b"content-length: ");
         out.extend_from_slice(body_len.to_string().as_bytes());
         out.extend_from_slice(b"\r\n");
@@ -229,10 +244,12 @@ pub fn serialise_response(conn: &Conn, keep_alive: bool) -> Vec<u8> {
 
     out.extend_from_slice(b"\r\n");
 
-    // Body.
+    // Fixed bodies are written inline with the head; a Stream body is
+    // pumped by the connection actor after this head goes on the wire.
     match &conn.resp_body {
-        RespBody::Empty    => {}
-        RespBody::Bytes(b) => out.extend_from_slice(b),
+        RespBody::Empty     => {}
+        RespBody::Bytes(b)  => out.extend_from_slice(b),
+        RespBody::Stream(_) => {}
     }
 
     out
@@ -366,6 +383,44 @@ mod tests {
         assert!(s.contains("connection: close"));
     }
 
+    #[test]
+    fn serialise_stream_http11_is_chunked_no_content_length() {
+        let (_tx, rx) = smarm::channel::<Vec<u8>>();
+        let conn = Conn::new().put_status(200).put_body(RespBody::from(rx));
+        let bytes = serialise_response(&conn, true);
+        let s = std::str::from_utf8(&bytes).unwrap();
+        assert!(s.contains("transfer-encoding: chunked"));
+        assert!(!s.contains("content-length"));
+        assert!(s.ends_with("\r\n\r\n")); // head only, no body bytes
+    }
+
+    #[test]
+    fn serialise_stream_strips_user_framing_headers() {
+        let (_tx, rx) = smarm::channel::<Vec<u8>>();
+        let conn = Conn::new().put_status(200)
+            .put_header("content-length", "999")
+            .put_header("transfer-encoding", "gzip")
+            .put_body(RespBody::from(rx));
+        let bytes = serialise_response(&conn, true);
+        let s = std::str::from_utf8(&bytes).unwrap();
+        assert!(!s.contains("content-length"));
+        assert!(!s.contains("gzip"));
+        assert!(s.contains("transfer-encoding: chunked"));
+    }
+
+    #[test]
+    fn serialise_stream_http10_no_te_and_closes() {
+        let (_tx, rx) = smarm::channel::<Vec<u8>>();
+        let mut conn = Conn::new().put_status(200).put_body(RespBody::from(rx));
+        conn.version = HttpVersion::Http10;
+        // The conn actor forces keep_alive=false for a 1.0 stream.
+        let bytes = serialise_response(&conn, false);
+        let s = std::str::from_utf8(&bytes).unwrap();
+        assert!(!s.contains("transfer-encoding"));
+        assert!(!s.contains("content-length"));
+        assert!(s.contains("connection: close"));
+    }
+
     #[test]
     fn serialise_user_content_length_is_respected() {
         let conn = Conn::new().put_status(200)

src/serve.rs

@@ -38,6 +38,9 @@ pub struct Config {
     pub max_header_count:   usize,
     pub read_buf_size:      usize,
     pub request_timeout:    Duration,
+    /// Per-write budget for response bytes (the fixed head+body write, and
+    /// each streamed chunk). See `ConnLimits::write_timeout`.
+    pub write_timeout:      Duration,
     pub max_body_bytes:     usize,
     /// How long a graceful shutdown waits for in-flight requests before
     /// force-stopping the remaining connections. Idle keep-alive
@@ -63,6 +66,7 @@ impl Config {
             max_header_count:   64,
             read_buf_size:      8 * 1024,
             request_timeout:    Duration::from_secs(30),
+            write_timeout:      Duration::from_secs(30),
             max_body_bytes:     16 * 1024 * 1024,
             drain_timeout:      Duration::from_secs(30),
             scheduler_threads:  None,
@@ -77,6 +81,7 @@ impl Config {
             max_body_bytes:     self.max_body_bytes,
             keep_alive_timeout: self.keep_alive_timeout,
             request_timeout:    self.request_timeout,
+            write_timeout:      self.write_timeout,
         }
     }
 }