feat(parser,conn): chunked request bodies (v0.3 chunk 2)

parser: Transfer-Encoding: chunked no longer 411s — it sets
ParsedHead.chunked and the conn actor decodes. Two hard rejections at
parse time, both Malformed/400: chunked together with Content-Length
(the request-smuggling ambiguity; RFC 7230 §3.3.3 permits rejection)
and chunked on HTTP/1.0 (TE is a 1.1 mechanism). ParseError::Unsupported
is now unconstructed; kept for future framings.

conn_actor: read_chunked_body decodes incrementally from buf[head_len..],
reading on the SAME request deadline the head came in under (a stalled
chunked body dies at request_timeout exactly like a stalled CL body).
Chunk extensions are ignored; trailers are consumed and discarded.
Bounds: decoded size capped at max_body_bytes -> 413 the moment the cap
would be crossed (the CL pre-check can't see a chunked body's size up
front); size lines capped at 128 bytes and the trailer section at 8 KiB
-> 400, so framing spam can't grow the read buffer unboundedly.

Returns (decoded, raw_consumed_past_head): the keep-alive drain at the
loop bottom now drops head_len + RAW framing length (not decoded length),
so pipelined requests behind a chunked body land exactly — covered by
the trailers+pipelining test.

Tests: 3 parser unit (flagging, CL+TE reject, 1.0 reject; the old
chunked-is-Unsupported test replaced) + 6 integration (decode with
extensions, trailers + pipelined next request, 413 over decoded cap,
400 malformed size line, 400 CL+TE on the wire, stalled chunked body
killed at request_timeout with silent close).

src/conn_actor.rs

@@ -133,6 +133,8 @@ pub fn run_connection(
         let _ = registry.cast(Cast::ConnBusy(me));
 
         // ----- 2. Read body. -----
+        // Content-Length pre-check only applies to fixed bodies; a chunked
+        // body is bounded incrementally by the decoder.
         let body_len = parsed.content_length.unwrap_or(0);
         if body_len > limits.max_body_bytes {
             let _ = write_all(
@@ -152,11 +154,41 @@ pub fn run_connection(
             }
         }
 
-        let body = match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) {
+        // `consumed_past_head`: how many RAW bytes of `buf` past the head
-            Ok(b) => b,
+        // this request's body occupied — for chunked bodies that is framing
-            // Timeout mid-body (and any other body io error) -> just close;
+        // included, NOT the decoded length. The keep-alive drain at the
-            // there's no point talking HTTP to a client this far gone.
+        // bottom of the loop must drop exactly this much to land on the
-            Err(_) => return,
+        // next pipelined request.
+        let (body, consumed_past_head) = if parsed.chunked {
+            match read_chunked_body(raw, &mut buf, parsed.head_len, &limits, request_deadline) {
+                Ok(ok) => ok,
+                Err(ChunkedBodyErr::TooLarge) => {
+                    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;
+                }
+                Err(ChunkedBodyErr::Malformed) => {
+                    let _ = write_all(
+                        raw,
+                        b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
+                        Instant::now() + limits.write_timeout,
+                    );
+                    return;
+                }
+                // Timeout mid-body (and any other io error) -> just close.
+                Err(ChunkedBodyErr::Io(_)) => return,
+            }
+        } else {
+            match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) {
+                Ok(b) => (b, body_len),
+                // Timeout mid-body (and any other body io error) -> just
+                // close; there's no point talking HTTP to a client this far
+                // gone.
+                Err(_) => return,
+            }
         };
 
         let keep_alive = parsed.keep_alive;
@@ -236,9 +268,9 @@ pub fn run_connection(
         // catch us, which is acceptable: drain means no new work.)
         let _ = registry.cast(Cast::ConnIdle(me));
 
-        // Drop the request bytes (head + body) from `buf`; anything past
+        // Drop the request bytes (head + raw body framing) from `buf`;
-        // them is the start of the next pipelined request.
+        // anything past them is the start of the next pipelined request.
-        let consumed = head_len + body_len;
+        let consumed = head_len + consumed_past_head;
         buf.drain(..consumed);
     }
 }
@@ -364,6 +396,135 @@ fn read_body(
     Ok(buf[head_len..head_len + body_len].to_vec())
 }
 
+// ---------------------------------------------------------------------------
+// read_chunked_body — incremental chunked transfer-decoding (request side).
+// ---------------------------------------------------------------------------
+//
+// Decodes `Transfer-Encoding: chunked` from `buf[head_len..]`, reading more
+// from the socket as needed on the SAME request deadline the head was read
+// under. Returns (decoded_body, raw_bytes_consumed_past_head) — the raw
+// count includes all framing and the trailer section, so the caller's
+// keep-alive drain lands exactly on the next pipelined request.
+//
+// Bounds: the DECODED size is capped at max_body_bytes (-> TooLarge/413);
+// a single size line (incl. chunk extensions, which are ignored) is capped
+// at MAX_SIZE_LINE and the trailer section at MAX_TRAILER_BYTES (-> 
+// Malformed/400) so framing spam can't grow `buf` unboundedly. Trailers
+// are consumed and discarded — nothing in the pipeline wants them yet.
+
+const MAX_SIZE_LINE:     usize = 128;
+const MAX_TRAILER_BYTES: usize = 8 * 1024;
+
+#[allow(dead_code)] // io::Error is captured for future logging
+enum ChunkedBodyErr {
+    Io(io::Error),
+    Malformed,
+    TooLarge,
+}
+
+fn read_chunked_body(
+    fd: RawFd,
+    buf: &mut Vec<u8>,
+    head_len: usize,
+    limits: &ConnLimits,
+    deadline: Instant,
+) -> Result<(Vec<u8>, usize), ChunkedBodyErr> {
+    // Ensure `buf` holds at least `until` bytes, reading on the request
+    // deadline. Io(TimedOut) on expiry, UnexpectedEof on early close.
+    fn fill_to(
+        fd: RawFd,
+        buf: &mut Vec<u8>,
+        until: usize,
+        deadline: Instant,
+    ) -> Result<(), ChunkedBodyErr> {
+        while buf.len() < until {
+            match read_some(fd, buf, 8 * 1024, deadline) {
+                Ok(0) => {
+                    return Err(ChunkedBodyErr::Io(io::Error::new(
+                        ErrorKind::UnexpectedEof,
+                        "client closed during chunked body",
+                    )))
+                }
+                Ok(_) => {}
+                Err(e) => return Err(ChunkedBodyErr::Io(e)),
+            }
+        }
+        Ok(())
+    }
+
+    // Find "\r\n" in buf[from..], reading more as needed; the line may be
+    // at most `max_line` bytes (terminator excluded). Returns the index of
+    // the '\r'.
+    fn find_crlf(
+        fd: RawFd,
+        buf: &mut Vec<u8>,
+        from: usize,
+        max_line: usize,
+        deadline: Instant,
+    ) -> Result<usize, ChunkedBodyErr> {
+        let mut scan = from;
+        loop {
+            while scan + 1 < buf.len() {
+                if buf[scan] == b'\r' && buf[scan + 1] == b'\n' {
+                    return Ok(scan);
+                }
+                scan += 1;
+                if scan - from > max_line {
+                    return Err(ChunkedBodyErr::Malformed);
+                }
+            }
+            fill_to(fd, buf, buf.len() + 1, deadline)?;
+        }
+    }
+
+    let mut pos = head_len;
+    let mut decoded: Vec<u8> = Vec::new();
+
+    loop {
+        // ----- size line: HEX[;extensions]\r\n -----
+        let line_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE, deadline)?;
+        let line = &buf[pos..line_end];
+        let size_str = match line.iter().position(|&b| b == b';') {
+            Some(i) => &line[..i], // chunk extensions: ignored
+            None    => line,
+        };
+        let size_str = std::str::from_utf8(size_str)
+            .map_err(|_| ChunkedBodyErr::Malformed)?
+            .trim();
+        let size = usize::from_str_radix(size_str, 16)
+            .map_err(|_| ChunkedBodyErr::Malformed)?;
+        pos = line_end + 2;
+
+        if size == 0 {
+            // ----- trailer section: zero or more header lines, then CRLF -----
+            let trailer_start = pos;
+            loop {
+                let t_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE.max(1024), deadline)?;
+                let empty = t_end == pos;
+                pos = t_end + 2;
+                if empty {
+                    return Ok((decoded, pos - head_len));
+                }
+                if pos - trailer_start > MAX_TRAILER_BYTES {
+                    return Err(ChunkedBodyErr::Malformed);
+                }
+            }
+        }
+
+        if decoded.len() + size > limits.max_body_bytes {
+            return Err(ChunkedBodyErr::TooLarge);
+        }
+
+        // ----- chunk payload + trailing CRLF -----
+        fill_to(fd, buf, pos + size + 2, deadline)?;
+        decoded.extend_from_slice(&buf[pos..pos + size]);
+        if &buf[pos + size..pos + size + 2] != b"\r\n" {
+            return Err(ChunkedBodyErr::Malformed);
+        }
+        pos += size + 2;
+    }
+}
+
 // ---------------------------------------------------------------------------
 // read_some — single epoll-park + read loop, bounded by a deadline.
 // ---------------------------------------------------------------------------

src/parser.rs

@@ -4,14 +4,13 @@
 //! copy, battle-tested). Body framing, keep-alive logic and response writing
 //! are ours.
 //!
-//! v1 body framing:
+//! Body framing:
 //!   - `Content-Length: N` — read exactly N bytes.
 //!   - No body header — empty body.
-//!   - `Transfer-Encoding: chunked` — deferred. Returns ParseError::Unsupported
+//!   - `Transfer-Encoding: chunked` (HTTP/1.1) — flagged in `ParsedHead`;
-//!     and the connection actor responds 411 Length Required + close.
+//!     the connection actor decodes incrementally (`read_chunked_body`).
-//!
+//!     Chunked + Content-Length together, or chunked on HTTP/1.0, is
-//! Keep it stupid simple. Chunked decoding lands when something actually
+//!     Malformed (request-smuggling ambiguity; RFC 7230 §3.3.3).
-//! requests it.
 
 use crate::conn::{Body, Conn, HeaderMap, HttpVersion, Method, RespBody};
 
@@ -30,8 +29,9 @@ pub enum ParseError {
     TooManyHeaders,
     /// `Content-Length` header could not be parsed as an integer.
     BadContentLength,
-    /// A wire feature we haven't implemented yet (e.g. chunked encoding).
+    /// A wire feature we haven't implemented. Currently unconstructed
-    /// Connection actor responds 411 + close.
+    /// (chunked decoding landed in v0.3); kept for future unsupported
+    /// framings. Connection actor responds 411 + close.
     Unsupported,
 }
 
@@ -50,6 +50,10 @@ pub struct ParsedHead {
     pub version:        HttpVersion,
     pub headers:        HeaderMap,
     pub content_length: Option<usize>,
+    /// `Transfer-Encoding: chunked` — the body is chunked-framed and the
+    /// connection actor decodes it (`read_chunked_body`). Mutually
+    /// exclusive with `content_length` (rejected as Malformed).
+    pub chunked:        bool,
     pub keep_alive:     bool,
     pub expect_100:     bool,
 }
@@ -130,7 +134,13 @@ pub fn parse_head(buf: &[u8], max_headers: usize) -> Result<ParsedHead, ParseErr
     }
 
     if chunked {
-        return Err(ParseError::Unsupported);
+        // Transfer-Encoding is an HTTP/1.1 mechanism; a 1.0 request
+        // carrying it is malformed. And a request carrying BOTH a
+        // Content-Length and TE: chunked is the classic request-smuggling
+        // ambiguity — RFC 7230 §3.3.3 lets a server reject it, and we do.
+        if version == HttpVersion::Http10 || content_length.is_some() {
+            return Err(ParseError::Malformed);
+        }
     }
 
     // Keep-alive logic, RFC 7230 §6.3:
@@ -149,6 +159,7 @@ pub fn parse_head(buf: &[u8], max_headers: usize) -> Result<ParsedHead, ParseErr
         version,
         headers,
         content_length,
+        chunked,
         keep_alive,
         expect_100,
     })
@@ -357,11 +368,28 @@ mod tests {
     }
 
     #[test]
-    fn parse_chunked_unsupported() {
+    fn parse_chunked_is_flagged() {
         let req = b"POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n";
+        let head = parse_head(req, 64).unwrap();
+        assert!(head.chunked);
+        assert_eq!(head.content_length, None);
+    }
+
+    #[test]
+    fn parse_chunked_plus_content_length_is_malformed() {
+        let req = b"POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nTransfer-Encoding: chunked\r\n\r\n";
         match parse_head(req, 64) {
-            Err(ParseError::Unsupported) => {}
+            Err(ParseError::Malformed) => {}
-            _ => panic!("expected Unsupported for chunked"),
+            _ => panic!("expected Malformed for CL + chunked"),
+        }
+    }
+
+    #[test]
+    fn parse_chunked_on_http10_is_malformed() {
+        let req = b"POST /a HTTP/1.0\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n";
+        match parse_head(req, 64) {
+            Err(ParseError::Malformed) => {}
+            _ => panic!("expected Malformed for chunked on 1.0"),
         }
     }
 

tests/integration.rs

@@ -772,3 +772,136 @@ fn stalled_reader_killed_at_write_timeout() {
         .recv_timeout(Duration::from_secs(10))
         .expect("producer never observed conn death: write_timeout not enforced");
 }
+
+// ---------------------------------------------------------------------------
+// Chunked request bodies (v0.3 chunk 2)
+// ---------------------------------------------------------------------------
+
+fn echo_pipeline() -> Pipeline {
+    Pipeline::new().plug(
+        Router::new()
+            .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello"))
+            .post("/echo", |c: Conn, _n: Next| {
+                let body = c.body.as_bytes().to_vec();
+                c.put_status(200).put_body(body)
+            }),
+    )
+}
+
+/// A chunked request body (with chunk extensions, which must be ignored)
+/// is decoded before the pipeline runs; the handler sees the joined bytes.
+#[test]
+fn chunked_request_body_is_decoded() {
+    let port = spawn_server(echo_pipeline());
+    let resp = send_request(
+        port,
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n\
+          5\r\nhello\r\n6;ext=1\r\n world\r\n0\r\n\r\n",
+    );
+    assert_eq!(http_status(&resp), 200);
+    assert_eq!(http_body(&resp), b"hello world");
+}
+
+/// Trailer headers after the 0-chunk are consumed and discarded; the
+/// connection stays usable (keep-alive across a chunked request, with the
+/// next request pipelined in the same packet — exercises the raw-consumed
+/// accounting).
+#[test]
+fn chunked_request_trailers_and_pipelining() {
+    let port = spawn_server(echo_pipeline());
+    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
+    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    s.write_all(
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
+          3\r\nabc\r\n0\r\nx-trailer: ignored\r\n\r\n\
+          GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
+    )
+    .unwrap();
+    let mut resp = Vec::new();
+    s.read_to_end(&mut resp).unwrap();
+    let text = String::from_utf8_lossy(&resp);
+    // Two responses back to back: the echo, then hello.
+    assert!(text.contains("abc"), "echo body missing: {text}");
+    let second = &text[text.find("abc").unwrap()..];
+    assert!(second.contains("hello"), "pipelined second response missing: {text}");
+}
+
+/// A chunked body whose decoded size exceeds max_body_bytes is rejected
+/// with 413 the moment the limit would be crossed.
+#[test]
+fn chunked_request_over_limit_413() {
+    let port = free_port();
+    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
+    let pipe = echo_pipeline();
+    std::thread::spawn(move || {
+        let cfg = Config {
+            listener_pool:     2,
+            scheduler_threads: Some(2),
+            max_body_bytes:    8, // tiny
+            ..Config::new(addr)
+        };
+        serve_with(cfg, pipe).unwrap();
+    });
+    for _ in 0..50 {
+        if TcpStream::connect(addr).is_ok() { break; }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    let mut s = TcpStream::connect(addr).unwrap();
+    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    s.write_all(
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
+          9\r\n123456789\r\n0\r\n\r\n",
+    )
+    .unwrap();
+    let mut resp = Vec::new();
+    s.read_to_end(&mut resp).unwrap();
+    assert_eq!(http_status(&resp), 413);
+}
+
+/// Garbage where a chunk-size line should be is 400.
+#[test]
+fn chunked_request_malformed_size_400() {
+    let port = spawn_server(echo_pipeline());
+    let resp = send_request(
+        port,
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
+          zz\r\nhello\r\n0\r\n\r\n",
+    );
+    assert_eq!(http_status(&resp), 400);
+}
+
+/// Content-Length + Transfer-Encoding: chunked together is the smuggling
+/// ambiguity: rejected 400 at parse time.
+#[test]
+fn chunked_plus_content_length_400() {
+    let port = spawn_server(echo_pipeline());
+    let resp = send_request(
+        port,
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nTransfer-Encoding: chunked\r\n\r\n\
+          5\r\nhello\r\n0\r\n\r\n",
+    );
+    assert_eq!(http_status(&resp), 400);
+}
+
+/// A chunked body that stalls mid-stream is killed by the request
+/// deadline: the connection just closes (no response owed mid-body).
+#[test]
+fn chunked_request_stall_killed_at_request_timeout() {
+    let port = spawn_server_with_timeouts(
+        echo_pipeline(),
+        Duration::from_secs(30),
+        Duration::from_millis(400), // request_timeout
+    );
+    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
+    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    // Head + first chunk, then stall forever before the 0-chunk.
+    s.write_all(
+        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n",
+    )
+    .unwrap();
+    let start = std::time::Instant::now();
+    let mut resp = Vec::new();
+    s.read_to_end(&mut resp).unwrap(); // server closes; EOF
+    assert!(resp.is_empty(), "expected silent close, got: {:?}", String::from_utf8_lossy(&resp));
+    assert!(start.elapsed() < Duration::from_secs(3), "close took too long");
+}