diff --git a/README.md b/README.md index caf7e5a..ec96304 100644 --- a/README.md +++ b/README.md @@ -132,7 +132,8 @@ let cfg = Config { listener_pool: 2, // Supervised accept-loop actors scheduler_threads: Some(2), // smarm worker threads (None = one per CPU) keep_alive_timeout: Duration::from_secs(60), // Idle budget between requests - request_timeout: Duration::from_secs(30), // Whole-request read deadline + request_timeout: Duration::from_secs(30), // Whole-request READ deadline + write_timeout: Duration::from_secs(30), // Per-write response budget max_header_count: 64, read_buf_size: 8 * 1024, max_body_bytes: 1 << 20, @@ -152,7 +153,14 @@ serve_with(cfg, pipeline).unwrap(); until its head and body are fully read. Expiry mid-head gets a best-effort `408 Request Timeout`; expiry mid-body just closes. Deadlines are absolute instants, so a trickling (slowloris-style) client can't - reset its budget by sending one byte at a time. + reset its budget by sending one byte at a time. Covers the **read phase + only** — a streaming response may legitimately outlive any whole-request + clock. +- `write_timeout` — a per-write budget for response bytes: the fixed + head+body write, and *each* streamed chunk, must complete within it. A + client that stops reading is dropped once its socket buffer fills and a + write stalls past the budget (the write-side twin of slowloris). Streamed + chunks each get a fresh budget; a stream as a whole has no deadline. ### Graceful Shutdown @@ -185,6 +193,62 @@ If every `Handle` is dropped, shutdown can never be signalled and the server runs forever — exactly `serve_with`'s semantics (it does this internally). +### Streaming Responses + +A handler can return a body it doesn't have yet: hand urus the read end of +a channel and feed it from a producer actor (the conn actor owns the +socket and all write deadlines; the producer never touches the fd): + +```rust +use urus::{Conn, Next}; + +|c: Conn, _n: Next| { + let (tx, rx) = smarm::channel::>(); + smarm::spawn(move || { + for part in ["chunk one", "chunk two"] { + if tx.send(part.as_bytes().to_vec()).is_err() { + return; // connection died — stop producing + } + } + // tx drops: end of stream. + }); + c.put_status(200).put_body(rx) +} +``` + +On HTTP/1.1 the response uses `Transfer-Encoding: chunked` and the +connection is reusable afterwards; on HTTP/1.0 (no chunked framing) bytes +go out raw and the connection closes to delimit the body. End of stream is +signalled by dropping every `Sender`. The producer contract: a `send` +error means the connection is gone — exit. Chunked **request** bodies are +decoded transparently; handlers see `conn.body` either way. + +### Server-Sent Events + +`Conn::sse()` is the streaming body with the SSE headers and a heartbeat +wired up: + +```rust +|c: Conn, _n: Next| { + let (c, events) = c.sse(); // or c.sse_with_heartbeat(interval) + smarm::spawn(move || { + loop { + if events.send("tick", "data").is_err() { + return; // SseClosed: client gone or server draining + } + smarm::sleep(std::time::Duration::from_secs(1)); + } + }); + c +} +``` + +While the producer is silent the connection actor emits `: keep-alive` +comment chunks every `HEARTBEAT_INTERVAL` (15s default). Liveness comes +from that ping: a vanished client is detected when a write stalls past +`write_timeout`. No request clock applies to an open SSE stream, and a +graceful shutdown force-stops it at the drain deadline. + ### Named Actors For introspection and debugging, the server registers itself in smarm's @@ -262,9 +326,9 @@ This avoids the complexity of async ecosystems while maintaining full concurrenc ## Roadmap See [`ROADMAP.md`](ROADMAP.md). Summary: v0.2 (supervised listener pool, -graceful shutdown, enforced timeouts, registry names) is done; next up are -streaming bodies + SSE (v0.3), WebSocket (v0.4), PubSub (v0.5), and -Phoenix-style channels (v0.6). +graceful shutdown, enforced timeouts, registry names) and v0.3 (streaming +bodies, chunked requests, SSE) are done; next up are WebSocket (v0.4), +PubSub (v0.5), and Phoenix-style channels (v0.6). Refer to `urus-spec.md` and `urus-v1-build-notes.md` in the artifact persistence for the original design and implementation notes. diff --git a/ROADMAP.md b/ROADMAP.md index 7ffe8c4..0ff45ac 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -119,22 +119,43 @@ panicking listener restarts under load without dropped accepts (test each). --- -## v0.3 — Streaming bodies + SSE +## v0.3 — Streaming bodies + SSE ✅ (landed 2026-06, chunks 1-3) The slipped v1 goal, and the WebSocket prerequisite (an upgraded connection is "a response that never ends"). -1. `RespBody::Stream(...)` — handler hands back a writer callback or a - `Receiver>`; conn actor switches to chunked transfer-encoding - (HTTP/1.1) and pumps until the stream closes. Keep-alive after a - completed chunked response. -2. Chunked **request** bodies (`parser.rs` currently returns 411) — same - cycle, the codec knowledge is fresh. -3. SSE sugar: `Conn::sse()` → an `EventSender` (`.send(event, data)`), - sets `content-type: text/event-stream`, disables buffering, heartbeats - via comment lines on a timer. An SSE conn is just a Stream body with - the per-read deadline disabled (or set to the heartbeat interval) — - liveness comes from the keep-alive ping, not `request_timeout`. +1. `RespBody::Stream(StreamBody)` ✅ — **decided: pull** (user call at the + v0.3 design fork): the handler hands back a `smarm::Receiver>` + and the producer is an actor the handler spawned; the conn actor pumps + (`pump_stream`) and keeps sole ownership of the socket and write + deadlines. The recv() park between chunks is stoppable by the draining + registry, so infinite streams die at the drain deadline with no + polling. HTTP/1.1: chunked TE, keep-alive after the terminator; + HTTP/1.0: raw bytes, forced close, EOF-delimited. End of stream = + every Sender dropped = 0-chunk. Timeout split (handoff Q3, as + proposed): `request_timeout` covers the READ phase only; new + `Config.write_timeout` (default 30s) bounds every response write — + fresh budget per streamed chunk, and the fixed-body + error-response + writes went deadline-bounded too (write-side slowloris closed). +2. Chunked **request** bodies ✅ — `read_chunked_body` decodes + incrementally on the request deadline; extensions ignored, trailers + consumed + discarded, decoded size capped at max_body_bytes (413), + size-line/trailer spam capped (400). CL + TE: chunked, or chunked on + HTTP/1.0, rejected 400 at parse (smuggling ambiguity). Keep-alive + drain accounts RAW framing length, so pipelined requests behind a + chunked body land exactly. +3. SSE sugar ✅ — `Conn::sse()` / `sse_with_heartbeat` → `EventSender` + (`send(event, data)` / `data` / `comment`, Clone, `Err(SseClosed)` + once the stream is gone). Heartbeat lives in the pump as leaned: with + `StreamBody.heartbeat` set, the chunk wait is `recv_timeout(interval)` + and expiry writes a `: keep-alive` comment chunk. Liveness from the + ping (dead client = heartbeat write stalls past write_timeout), not + from any request clock. + +**Exit criteria (verified):** chunked round-trips both directions with +keep-alive and pipelining intact; an infinite SSE stream heartbeats while +silent, is killed by write_timeout when the client stops reading, and is +force-stopped at the drain deadline on shutdown; full suite green. --- diff --git a/examples/crud.rs b/examples/crud.rs index 02cedea..55de4ae 100644 --- a/examples/crud.rs +++ b/examples/crud.rs @@ -204,6 +204,28 @@ fn parse_id(s: &str) -> Option { // Handlers // --------------------------------------------------------------------------- +/// SSE demo: `curl -N localhost:8080/ticker` streams a tick every second +/// (with `: keep-alive` comments if it ever goes quiet). The producer +/// exits on SseClosed (client gone / write timeout / shutdown drain) or +/// when the example is shutting down. +fn ticker(conn: Conn, _next: Next) -> Conn { + let (conn, events) = conn.sse(); + smarm::spawn(move || { + let mut n: u64 = 0; + loop { + if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) { + return; // dropping `events` ends the stream cleanly + } + if events.send("tick", &n.to_string()).is_err() { + return; // SseClosed + } + n += 1; + smarm::sleep(std::time::Duration::from_secs(1)); + } + }); + conn +} + fn list(conn: Conn, _next: Next) -> Conn { let (tx, rx) = channel::<(u16, Vec)>(); store().send(Request::List { reply: tx }).ok(); @@ -278,7 +300,8 @@ fn main() { .post( "/users", create) .get( "/users/:id", get_one) .put( "/users/:id", update) - .delete("/users/:id", delete), + .delete("/users/:id", delete) + .get( "/ticker", ticker), ); let cfg = Config::new("127.0.0.1:8080".parse().unwrap());