diff --git a/.gitignore b/.gitignore index a9d37c5..01eb4b6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ target Cargo.lock +smarm_trace.json diff --git a/examples/crud.rs b/examples/crud.rs index 2e64413..02cedea 100644 --- a/examples/crud.rs +++ b/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) { 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> = 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 { STORE_TX.get_or_init(|| { let (tx, rx) = channel::(); @@ -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"); } diff --git a/src/conn_actor.rs b/src/conn_actor.rs index fd61edf..b940e77 100644 --- a/src/conn_actor.rs +++ b/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, limits: &ConnLimits, -) -> Result { +) -> 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 = 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, head_len: usize, body_len: usize, + deadline: Instant, ) -> io::Result> { // 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, chunk: usize) -> io::Result { +fn read_some( + fd: RawFd, + buf: &mut Vec, + chunk: usize, + deadline: Instant, +) -> io::Result { // 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, chunk: usize) -> io::Result { } } +// --------------------------------------------------------------------------- +// 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. // --------------------------------------------------------------------------- diff --git a/src/serve.rs b/src/serve.rs index df98f53..26d6c49 100644 --- a/src/serve.rs +++ b/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, } } } diff --git a/tests/integration.rs b/tests/integration.rs index 2c946bf..ad4bba7 100644 --- a/tests/integration.rs +++ b/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 { + 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); +}