feat(conn): keep-alive and per-request read timeouts (v0.2 chunk 3)

Two wall-clock budgets in the conn read path, both via
smarm::wait_readable_timeout:

- keep_alive_timeout: idle budget between requests — how long we park
  waiting for the FIRST byte of a request (including the first request on
  a fresh connection). Expiry closes silently; nothing is owed to a
  client that isn't talking.
- request_timeout: per-request budget as an Instant deadline from the
  first byte of a request until the request (head + body) is fully read.
  Pipelined leftovers in the buffer at cycle start count as a started
  request. The deadline is returned from read_head and threaded into
  read_body so head and body share one budget. Pipeline run time is NOT
  covered. Each read wait uses whichever budget is currently active;
  deadlines are Instants, so EAGAIN retries don't reset the clock.

Expiry mid-head gets a best-effort 408 via try_write_once: a single
non-parking write syscall — a client that stalls its read side must not
defeat the timeout by making the 408 write park forever. Expiry mid-body
just closes.

examples/crud.rs gains stdin-Enter graceful shutdown (plain OS thread on
read_line -> handle.shutdown(); no signal crate). Doing so surfaced a
pattern worth knowing: crud's lazily-spawned store actor parked forever
in recv(), which blocks smarm's AllDone, so serve_with_shutdown never
returned (gdb: scheduler idle in poll_wake, store the only live actor).
It can't be messaged awake from the stdin thread either — a cross-thread
send's unpark is a no-op without runtime TLS, the same limitation behind
SHUTDOWN_POLL. Fix: store_loop recv_timeout(250ms) + a SHUTTING_DOWN
AtomicBool set by the stdin thread before handle.shutdown(). This poll
dies with the cross-thread-unpark limitation; recorded in smarm docs.

Tests: idle keep-alive conn reaped at a small keep_alive_timeout;
slowloris partial-head stall killed at request_timeout with the 408
observed. Timeout+shutdown subset hammered 30x clean; full suite 3x;
smarm-trace build and suite clean.

.gitignore

@@ -1,2 +1,3 @@
 target
 Cargo.lock
+smarm_trace.json

examples/crud.rs

@@ -25,7 +25,7 @@
 use serde::{Deserialize, Serialize};
 use smarm::{channel, Sender};
 use std::sync::OnceLock;
-use urus::{serve_with, Config, Conn, Next, Pipeline, Router};
+use urus::{serve_with_shutdown, shutdown_handle, Config, Conn, Next, Pipeline, Router};
 
 // ---------------------------------------------------------------------------
 // Domain
@@ -78,8 +78,19 @@ fn store_loop(rx: smarm::Receiver<Request>) {
     let mut next_id: u64 = users.iter().map(|u| u.id).max().unwrap_or(0) + 1;
 
     loop {
-        let req = match rx.recv() {
-            Ok(r)  => r,
+        // recv with a timeout rather than a bare recv: the store must be
+        // stoppable at shutdown, but a cross-thread Sender::send (from the
+        // stdin thread) can't wake a parked actor — same smarm limitation
+        // that motivates urus's SHUTDOWN_POLL. So we wake on our own timer
+        // and poll the flag. This poll dies with that limitation too.
+        let req = match rx.recv_timeout(std::time::Duration::from_millis(250)) {
+            Ok(r) => r,
+            Err(smarm::channel::RecvTimeoutError::Timeout) => {
+                if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
+                    return;
+                }
+                continue;
+            }
             Err(_) => return, // all senders dropped
         };
         match req {
@@ -165,6 +176,12 @@ fn persist(users: &[User]) {
 
 static STORE_TX: OnceLock<Sender<Request>> = OnceLock::new();
 
+// Set by the stdin thread at shutdown; the store actor polls it (see
+// store_loop). An always-on app actor that never returns would otherwise
+// block smarm's AllDone and keep serve_with_shutdown from returning.
+static SHUTTING_DOWN: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
+
 fn store() -> &'static Sender<Request> {
     STORE_TX.get_or_init(|| {
         let (tx, rx) = channel::<Request>();
@@ -266,5 +283,19 @@ fn main() {
 
     let cfg = Config::new("127.0.0.1:8080".parse().unwrap());
     println!("urus-crud: DB at {DB_PATH}");
-    serve_with(cfg, pipeline).unwrap();
+    println!("urus-crud: listening on 127.0.0.1:8080 — press Enter to shut down");
+
+    // Graceful shutdown on stdin-Enter: a plain OS thread blocks on
+    // read_line and fires the handle. No signal handling crate needed.
+    let (handle, signal) = shutdown_handle();
+    std::thread::spawn(move || {
+        let mut line = String::new();
+        let _ = std::io::stdin().read_line(&mut line);
+        println!("urus-crud: shutting down (draining in-flight requests)…");
+        SHUTTING_DOWN.store(true, std::sync::atomic::Ordering::Relaxed);
+        handle.shutdown();
+    });
+
+    serve_with_shutdown(cfg, pipeline, signal).unwrap();
+    println!("urus-crud: bye");
 }

src/conn_actor.rs

@@ -23,7 +23,7 @@ use smarm::ServerRef;
 
 use std::io::{self, ErrorKind};
 use std::os::fd::RawFd;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 // ---------------------------------------------------------------------------
 // Limits
@@ -41,7 +41,16 @@ pub struct ConnLimits {
     /// Hard cap on Content-Length we'll accept. 16 MiB is enough for a CRUD
     /// example; configurable in `Config`.
     pub max_body_bytes:     usize,
+    /// Idle budget between requests: how long we'll park waiting for the
+    /// FIRST byte of a request (including the first request on a fresh
+    /// connection). Expiry closes the connection silently — nothing is
+    /// owed to a client that isn't talking.
     pub keep_alive_timeout: Duration,
+    /// Per-request wall-clock budget, measured from the first byte of a
+    /// 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.
+    pub request_timeout:    Duration,
 }
 
 impl Default for ConnLimits {
@@ -52,6 +61,7 @@ impl Default for ConnLimits {
             max_head_bytes:     64 * 1024,
             max_body_bytes:     16 * 1024 * 1024,
             keep_alive_timeout: Duration::from_secs(60),
+            request_timeout:    Duration::from_secs(30),
         }
     }
 }
@@ -83,14 +93,27 @@ pub fn run_connection(
         // ----- 1. Read until we have a full request head. -----
         // We are idle until a head parses: stoppable by a draining
         // registry while parked here.
-        let parsed = match read_head(raw, &mut buf, &limits) {
+        let (parsed, request_deadline) = match read_head(raw, &mut buf, &limits) {
             Ok(p) => p,
             Err(ReadHeadErr::ClientClosed) => {
                 // Clean EOF between requests (or before any request). Normal.
                 return;
             }
+            Err(ReadHeadErr::IdleTimeout) => {
+                // keep_alive_timeout expired waiting for the first byte of
+                // a request. Nothing is owed; close silently.
+                return;
+            }
+            Err(ReadHeadErr::RequestTimeout) => {
+                // request_timeout expired mid-head (slowloris and friends).
+                // Best-effort 408 WITHOUT parking on writability — a client
+                // that stalls reads must not defeat the timeout by making
+                // the 408 write park forever.
+                try_write_once(raw, b"HTTP/1.1 408 Request Timeout\r\ncontent-length: 0\r\nconnection: close\r\n\r\n");
+                return;
+            }
             Err(ReadHeadErr::Io(_)) => {
-                // Network error or timeout. Best-effort close; we're done.
+                // Network error. Best-effort close; we're done.
                 return;
             }
             Err(ReadHeadErr::Parse(e)) => {
@@ -116,8 +139,10 @@ pub fn run_connection(
             }
         }
 
-        let body = match read_body(raw, &mut buf, parsed.head_len, body_len) {
+        let body = match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) {
             Ok(b) => b,
+            // 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,
         };
 
@@ -194,6 +219,12 @@ pub fn run_connection(
 #[allow(dead_code)] // io::Error is captured for future logging
 enum ReadHeadErr {
     ClientClosed,
+    /// keep_alive_timeout expired while waiting for the first byte of a
+    /// request. Close silently.
+    IdleTimeout,
+    /// request_timeout expired after the request had started arriving.
+    /// Best-effort 408.
+    RequestTimeout,
     Io(io::Error),
     Parse(ParseError),
 }
@@ -201,17 +232,41 @@ enum ReadHeadErr {
 /// Read until `parse_head` succeeds or fails definitively. `buf` may already
 /// contain leftover bytes from a previous keep-alive cycle; we try to parse
 /// those before reading more from the socket.
+///
+/// Two wall-clock budgets govern the waits (each wait uses whichever budget
+/// is currently active):
+///
+/// - while `buf` is empty and nothing has arrived, we are *idle* and the
+///   wait is bounded by `keep_alive_timeout`;
+/// - the instant the request has started (first byte read, or pipelined
+///   bytes already in `buf` at entry), the *request* clock starts: an
+///   `Instant` deadline of `request_timeout` from that moment, which also
+///   covers body reads — it is returned alongside the parsed head so the
+///   caller can thread it into `read_body`.
 fn read_head(
     fd: RawFd,
     buf: &mut Vec<u8>,
     limits: &ConnLimits,
-) -> Result<parser::ParsedHead, ReadHeadErr> {
+) -> Result<(parser::ParsedHead, Instant), ReadHeadErr> {
+    let entry = Instant::now();
+    let idle_deadline = entry + limits.keep_alive_timeout;
+    // Pipelined leftovers count as a started request.
+    let mut request_deadline: Option<Instant> = if buf.is_empty() {
+        None
+    } else {
+        Some(entry + limits.request_timeout)
+    };
+
     loop {
         // Try to parse what we already have. On the first iteration of a
         // fresh keep-alive cycle, `buf` may already hold the next request.
         if !buf.is_empty() {
             match parser::parse_head(buf, limits.max_headers) {
-                Ok(h)  => return Ok(h),
+                Ok(h)  => {
+                    let deadline = request_deadline
+                        .unwrap_or_else(|| Instant::now() + limits.request_timeout);
+                    return Ok((h, deadline));
+                }
                 Err(ParseError::Incomplete) => {} // need more bytes
                 Err(e) => return Err(ReadHeadErr::Parse(e)),
             }
@@ -221,10 +276,24 @@ fn read_head(
             return Err(ReadHeadErr::Parse(ParseError::TooManyHeaders));
         }
 
-        // Read more.
-        match read_some(fd, buf, limits.initial_read_buf) {
+        // Read more, bounded by whichever budget is active.
+        let deadline = request_deadline.unwrap_or(idle_deadline);
+        match read_some(fd, buf, limits.initial_read_buf, deadline) {
             Ok(0)  => return Err(ReadHeadErr::ClientClosed),
-            Ok(_)  => continue,
+            Ok(_)  => {
+                if request_deadline.is_none() {
+                    // First byte(s) of this request: the request clock
+                    // starts now.
+                    request_deadline = Some(Instant::now() + limits.request_timeout);
+                }
+            }
+            Err(e) if e.kind() == ErrorKind::TimedOut => {
+                return Err(if request_deadline.is_some() {
+                    ReadHeadErr::RequestTimeout
+                } else {
+                    ReadHeadErr::IdleTimeout
+                });
+            }
             Err(e) => return Err(ReadHeadErr::Io(e)),
         }
     }
@@ -239,6 +308,7 @@ fn read_body(
     buf: &mut Vec<u8>,
     head_len: usize,
     body_len: usize,
+    deadline: Instant,
 ) -> io::Result<Vec<u8>> {
     // Bytes already in `buf` past the head belong to the body.
     let already = buf.len().saturating_sub(head_len);
@@ -250,10 +320,11 @@ fn read_body(
         return Ok(buf[head_len..head_len + body_len].to_vec());
     }
 
-    // Read until we have the rest.
+    // Read until we have the rest, on the same request budget that the
+    // head was read under.
     let mut total_read = already;
     while total_read < body_len {
-        match read_some(fd, buf, 8 * 1024) {
+        match read_some(fd, buf, 8 * 1024, deadline) {
             Ok(0)  => return Err(io::Error::new(ErrorKind::UnexpectedEof, "client closed during body")),
             Ok(n)  => total_read += n,
             Err(e) => return Err(e),
@@ -263,18 +334,31 @@ fn read_body(
 }
 
 // ---------------------------------------------------------------------------
-// read_some — single epoll-park + read loop.
+// read_some — single epoll-park + read loop, bounded by a deadline.
 // ---------------------------------------------------------------------------
 //
-// Appends what it reads onto `buf`. Returns bytes read, 0 for EOF, or the
-// last io error.
+// Appends what it reads onto `buf`. Returns bytes read, 0 for EOF,
+// `ErrorKind::TimedOut` when `deadline` passes before the fd turns
+// readable, or the last io error.
 
-fn read_some(fd: RawFd, buf: &mut Vec<u8>, chunk: usize) -> io::Result<usize> {
+fn read_some(
+    fd: RawFd,
+    buf: &mut Vec<u8>,
+    chunk: usize,
+    deadline: Instant,
+) -> io::Result<usize> {
     // Loop to absorb EAGAIN: a readable wakeup followed by EAGAIN is
     // possible (signal race, etc). Re-park and retry rather than returning
-    // 0 (which would be confused with EOF by callers).
+    // 0 (which would be confused with EOF by callers). The deadline is an
+    // Instant, so spurious wakes don't reset the budget.
     loop {
-        smarm::wait_readable(fd)?;
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(io::Error::new(ErrorKind::TimedOut, "read deadline elapsed"));
+        }
+        if !smarm::wait_readable_timeout(fd, remaining)? {
+            return Err(io::Error::new(ErrorKind::TimedOut, "read deadline elapsed"));
+        }
 
         let start = buf.len();
         buf.resize(start + chunk, 0);
@@ -298,6 +382,22 @@ fn read_some(fd: RawFd, buf: &mut Vec<u8>, chunk: usize) -> io::Result<usize> {
     }
 }
 
+// ---------------------------------------------------------------------------
+// try_write_once — single non-parking write attempt, result ignored.
+// ---------------------------------------------------------------------------
+//
+// For best-effort farewells (the 408) to clients we've decided to drop:
+// one non-blocking write syscall, no wait_writable park. A client that
+// stalls its read side must not be able to keep this actor alive past its
+// own timeout. The socket send buffer almost always has room for a
+// one-liner, so in practice the 408 lands.
+
+fn try_write_once(fd: RawFd, buf: &[u8]) {
+    unsafe {
+        let _ = libc::write(fd, buf.as_ptr() as *const _, buf.len());
+    }
+}
+
 // ---------------------------------------------------------------------------
 // write_all — robust write loop.
 // ---------------------------------------------------------------------------

src/serve.rs

@@ -76,6 +76,7 @@ impl Config {
             max_head_bytes:     64 * 1024,
             max_body_bytes:     self.max_body_bytes,
             keep_alive_timeout: self.keep_alive_timeout,
+            request_timeout:    self.request_timeout,
         }
     }
 }

tests/integration.rs

@@ -425,3 +425,115 @@ fn shutdown_force_stops_at_drain_deadline() {
         .recv_timeout(Duration::from_secs(5))
         .expect("serve did not return after force-stop deadline");
 }
+
+// ---------------------------------------------------------------------------
+// Timeouts (v0.2 chunk 3)
+// ---------------------------------------------------------------------------
+
+fn spawn_server_with_timeouts(
+    pipeline: Pipeline,
+    keep_alive: Duration,
+    request: Duration,
+) -> u16 {
+    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),
+            keep_alive_timeout: keep_alive,
+            request_timeout:    request,
+            ..Config::new(addr)
+        };
+        serve_with(cfg, pipeline).unwrap();
+    });
+    for _ in 0..50 {
+        if TcpStream::connect(addr).is_ok() {
+            return port;
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    panic!("server didn't come up on {addr}");
+}
+
+/// Read from `s` until the response head is complete (double CRLF). Only
+/// suitable for responses with an empty body.
+fn read_response_head(s: &mut TcpStream) -> Vec<u8> {
+    let mut resp = Vec::new();
+    let mut byte = [0u8; 1];
+    while !resp.ends_with(b"\r\n\r\n") {
+        match s.read(&mut byte) {
+            Ok(0) => break,
+            Ok(_) => resp.push(byte[0]),
+            Err(e) => panic!("read failed mid-response: {e}"),
+        }
+    }
+    resp
+}
+
+/// An idle keep-alive connection is reaped once keep_alive_timeout passes
+/// with no next request: the server closes the socket (clean EOF on our
+/// side) well before the much larger request_timeout.
+#[test]
+fn idle_keepalive_reaped_at_keep_alive_timeout() {
+    let pipe = Pipeline::new().plug(
+        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
+    );
+    let port = spawn_server_with_timeouts(
+        pipe,
+        Duration::from_millis(300), // keep_alive_timeout under test
+        Duration::from_secs(10),    // request_timeout out of the way
+    );
+
+    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 / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); // keep-alive
+    let head = read_response_head(&mut s);
+    assert_eq!(http_status(&head), 200);
+
+    // Now go quiet. The server should close us at ~300ms; the 5s read
+    // timeout on our side is the failure detector.
+    let start = std::time::Instant::now();
+    let mut byte = [0u8; 1];
+    match s.read(&mut byte) {
+        Ok(0) => {} // clean EOF: reaped
+        Ok(n) => panic!("expected EOF, got {n} unexpected byte(s)"),
+        Err(e) => panic!("expected EOF, read errored: {e}"),
+    }
+    assert!(
+        start.elapsed() < Duration::from_secs(3),
+        "reap took {:?}, expected ~300ms", start.elapsed()
+    );
+}
+
+/// A slowloris client that sends a partial head and then stalls is killed
+/// at request_timeout with a best-effort 408, even though the (large)
+/// keep-alive budget hasn't expired.
+#[test]
+fn slowloris_partial_head_killed_at_request_timeout() {
+    let pipe = Pipeline::new().plug(
+        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
+    );
+    let port = spawn_server_with_timeouts(
+        pipe,
+        Duration::from_secs(10),    // keep_alive_timeout out of the way
+        Duration::from_millis(300), // request_timeout under test
+    );
+
+    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
+    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    // Partial head: request clock starts on these bytes, never completes.
+    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n").unwrap();
+
+    let start = std::time::Instant::now();
+    let mut resp = Vec::new();
+    s.read_to_end(&mut resp).expect("expected 408+EOF or EOF");
+    assert!(
+        start.elapsed() < Duration::from_secs(3),
+        "kill took {:?}, expected ~300ms", start.elapsed()
+    );
+    // The 408 is best-effort (single non-parking write), but with our read
+    // side live it should land.
+    assert!(!resp.is_empty(), "expected a best-effort 408 before close");
+    assert_eq!(http_status(&resp), 408);
+}