From 42a24647430a44fd00e101184c0634f754880ef3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 12 Jun 2026 04:53:27 +0000 Subject: [PATCH] =?UTF-8?q?feat(conn):=20streaming=20response=20bodies=20?= =?UTF-8?q?=E2=80=94=20RespBody::Stream=20+=20chunked=20TE=20(v0.3=20chunk?= =?UTF-8?q?=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Design decisions (per handoff Q1 user answer + Q3 proposal): - Producer shape is PULL: the handler returns a smarm Receiver> (RespBody::Stream / StreamBody, From>> 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 | 60 ++++++++++-- src/conn_actor.rs | 123 +++++++++++++++++++++--- src/lib.rs | 2 +- src/parser.rs | 67 ++++++++++++-- src/serve.rs | 5 + tests/integration.rs | 216 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 448 insertions(+), 25 deletions(-) diff --git a/src/conn.rs b/src/conn.rs index cecb5b7..2658c42 100644 --- a/src/conn.rs +++ b/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>` — 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), + 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>, +} + +impl StreamBody { + pub fn new(rx: smarm::Receiver>) -> 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 for RespBody { + fn from(s: StreamBody) -> Self { RespBody::Stream(s) } +} +impl From>> for RespBody { + fn from(rx: smarm::Receiver>) -> Self { + RespBody::Stream(StreamBody::new(rx)) + } +} diff --git a/src/conn_actor.rs b/src/conn_actor.rs index b940e77..70b6a35 100644 --- a/src/conn_actor.rs +++ b/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); } diff --git a/src/lib.rs b/src/lib.rs index 488efb2..e85d03e 100644 --- a/src/lib.rs +++ b/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::{ diff --git a/src/parser.rs b/src/parser.rs index 289e18f..7c93352 100644 --- a/src/parser.rs +++ b/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 { // 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 { 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 { 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 { 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::>(); + 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::>(); + 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::>(); + 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) diff --git a/src/serve.rs b/src/serve.rs index bd7c35f..9135937 100644 --- a/src/serve.rs +++ b/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, } } } diff --git a/tests/integration.rs b/tests/integration.rs index 6483a34..dfc06d8 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -556,3 +556,219 @@ fn slowloris_partial_head_killed_at_request_timeout() { assert!(!resp.is_empty(), "expected a best-effort 408 before close"); assert_eq!(http_status(&resp), 408); } + +// --------------------------------------------------------------------------- +// Streaming responses (v0.3 chunk 1) +// --------------------------------------------------------------------------- + +/// Router with a fixed route plus a 3-chunk streaming route. The producer +/// is a separate actor (the pull design): the handler spawns it, hands the +/// Receiver to urus, and returns. Dropping the Sender ends the stream. +fn streaming_pipeline(chunk_gap: Duration, chunks: usize) -> Pipeline { + Pipeline::new().plug( + Router::new() + .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello")) + .get("/stream", move |c: Conn, _n: Next| { + let (tx, rx) = smarm::channel::>(); + smarm::spawn(move || { + for i in 0..chunks { + if tx.send(format!("part{i}").into_bytes()).is_err() { + return; // conn died; stop producing + } + smarm::sleep(chunk_gap); + } + }); + c.put_status(200).put_body(rx) + }), + ) +} + +/// Read from `s` until the chunked-encoding terminator arrives, then +/// return (head, decoded_body). +fn read_chunked_response(s: &mut TcpStream) -> (String, Vec) { + let mut raw = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + match s.read(&mut buf) { + Ok(0) => panic!("connection closed before chunked terminator"), + Ok(n) => raw.extend_from_slice(&buf[..n]), + Err(e) => panic!("read failed mid-stream: {e}"), + } + if raw.windows(5).any(|w| w == b"0\r\n\r\n") { + break; + } + } + let head_end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4; + let head = String::from_utf8_lossy(&raw[..head_end]).to_string(); + + // Decode the chunked body. + let mut body = Vec::new(); + let mut rest = &raw[head_end..]; + loop { + let line_end = rest.windows(2).position(|w| w == b"\r\n").unwrap(); + let size = usize::from_str_radix( + std::str::from_utf8(&rest[..line_end]).unwrap().trim(), + 16, + ) + .unwrap(); + rest = &rest[line_end + 2..]; + if size == 0 { + break; + } + body.extend_from_slice(&rest[..size]); + assert_eq!(&rest[size..size + 2], b"\r\n", "chunk not CRLF-terminated"); + rest = &rest[size + 2..]; + } + (head, body) +} + +/// A streaming handler produces a chunked HTTP/1.1 response whose decoded +/// body is the concatenation of the producer's sends, with no +/// content-length on the wire. +#[test] +fn streaming_response_chunked() { + let port = spawn_server(streaming_pipeline(Duration::from_millis(10), 3)); + 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"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + + let (head, body) = read_chunked_response(&mut s); + assert!(head.starts_with("HTTP/1.1 200 OK\r\n"), "head: {head}"); + assert!(head.contains("transfer-encoding: chunked"), "head: {head}"); + assert!(!head.contains("content-length"), "head: {head}"); + assert_eq!(body, b"part0part1part2"); +} + +/// The connection survives a completed chunked response: a second request +/// on the same socket gets a normal fixed response. +#[test] +fn keep_alive_after_chunked_response() { + let port = spawn_server(streaming_pipeline(Duration::from_millis(5), 3)); + 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"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + let (_head, body) = read_chunked_response(&mut s); + assert_eq!(body, b"part0part1part2"); + + // Same socket, second request. + s.write_all(b"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(); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"hello"); +} + +/// HTTP/1.0 has no chunked framing: a stream body goes out raw, the +/// response carries `connection: close` and no transfer-encoding, and EOF +/// delimits the body. +#[test] +fn http10_stream_is_eof_delimited() { + let port = spawn_server(streaming_pipeline(Duration::from_millis(5), 3)); + 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"GET /stream HTTP/1.0\r\nHost: x\r\n\r\n").unwrap(); + + let mut resp = Vec::new(); + s.read_to_end(&mut resp).unwrap(); // EOF = end of body + let head = String::from_utf8_lossy(&resp[..]).to_string(); + assert!(head.starts_with("HTTP/1.0 200 OK\r\n"), "head: {head}"); + assert!(!head.contains("transfer-encoding"), "head: {head}"); + assert!(head.contains("connection: close"), "head: {head}"); + assert_eq!(http_body(&resp), b"part0part1part2"); +} + +/// An infinite stream is an in-flight request: graceful shutdown waits for +/// the drain deadline, then force-stops the conn actor parked in the pump's +/// recv(). serve returns; the client sees EOF; the orphaned producer's next +/// send fails (closed channel) and it exits, letting the runtime wind down. +#[test] +fn shutdown_force_stops_active_stream() { + let pipe = Pipeline::new().plug(Router::new().get( + "/infinite", + |c: Conn, _n: Next| { + let (tx, rx) = smarm::channel::>(); + smarm::spawn(move || loop { + if tx.send(b"tick".to_vec()).is_err() { + return; + } + smarm::sleep(Duration::from_millis(20)); + }); + c.put_status(200).put_body(rx) + }, + )); + let (port, handle, done_rx) = + spawn_server_with_handle(pipe, Duration::from_millis(300)); + + 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"GET /infinite HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + + // Confirm the stream is live before shutting down. + let mut buf = [0u8; 1024]; + let n = s.read(&mut buf).unwrap(); + assert!(n > 0); + + handle.shutdown(); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("serve did not return: streaming conn not force-stopped at drain deadline"); + + // Client side drains to EOF. + let mut rest = Vec::new(); + let _ = s.read_to_end(&mut rest); // EOF or reset; both fine +} + +/// A client that stops reading mid-stream is dropped once a chunk write +/// stalls past write_timeout; the producer observes the closed channel. +#[test] +fn stalled_reader_killed_at_write_timeout() { + let (dead_tx, dead_rx) = std::sync::mpsc::channel::<()>(); + let pipe = Pipeline::new().plug(Router::new().get( + "/firehose", + move |c: Conn, _n: Next| { + let (tx, rx) = smarm::channel::>(); + let dead_tx = dead_tx.clone(); + smarm::spawn(move || loop { + if tx.send(vec![0u8; 64 * 1024]).is_err() { + let _ = dead_tx.send(()); // conn actor dropped the Receiver + return; + } + smarm::sleep(Duration::from_millis(1)); + }); + c.put_status(200).put_body(rx) + }, + )); + + let port = free_port(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + std::thread::spawn(move || { + let cfg = Config { + listener_pool: 2, + scheduler_threads: Some(2), + write_timeout: Duration::from_millis(300), + ..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.write_all(b"GET /firehose HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + // Read a little to confirm liveness, then stall: never read again. + let mut buf = [0u8; 1024]; + let _ = s.read(&mut buf).unwrap(); + + // Socket buffers fill, a chunk write stalls, write_timeout (300ms) + // expires, the conn actor exits, the Receiver drops, the producer's + // send fails. Generous wall budget for slow CI. + dead_rx + .recv_timeout(Duration::from_secs(10)) + .expect("producer never observed conn death: write_timeout not enforced"); +}