commit 3b6c46621039fbff0b5eb51cdcb7769839e1705f Author: Mark Kalsbeek Date: Tue May 26 23:16:45 2026 +0200 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a9d37c5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target +Cargo.lock diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..d557371 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "urus" +version = "0.1.0" +edition = "2021" +rust-version = "1.95" +description = "Cowboy/bandit-style HTTP library for the smarm actor runtime" + +[dependencies] +smarm = { path = "../smarm" } +httparse = "1.9" +libc = "0.2" + +[features] +smarm-trace = ["smarm/smarm-trace"] + +[dev-dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" + +[profile.dev] +panic = "unwind" + +[profile.release] +panic = "unwind" +lto = "thin" +codegen-units = 1 + +[[example]] +name = "crud" +path = "examples/crud.rs" diff --git a/examples/crud.rs b/examples/crud.rs new file mode 100644 index 0000000..2e64413 --- /dev/null +++ b/examples/crud.rs @@ -0,0 +1,270 @@ +//! CRUD example: a tiny user database with JSON persistence. +//! +//! Demonstrates urus and the actor model together: +//! - The pipeline is shared (Arc) across all connection actors. +//! - Handlers do NOT take a lock or share mutable state directly. +//! - A single "store" actor owns the data; handlers send it a request +//! via a channel and block on the reply. Serialization is structural — +//! the store processes one request at a time, no Mutex needed. +//! - On every mutating request the store writes the JSON file. Read +//! requests don't touch disk. +//! +//! Endpoints: +//! GET /users — list +//! POST /users — create +//! GET /users/:id — fetch +//! PUT /users/:id — replace +//! DELETE /users/:id — delete +//! +//! Try it: +//! cargo run --example crud +//! curl -s http://localhost:8080/users +//! curl -s -X POST -d '{"name":"alice","email":"a@x"}' http://localhost:8080/users +//! curl -s http://localhost:8080/users/1 + +use serde::{Deserialize, Serialize}; +use smarm::{channel, Sender}; +use std::sync::OnceLock; +use urus::{serve_with, Config, Conn, Next, Pipeline, Router}; + +// --------------------------------------------------------------------------- +// Domain +// --------------------------------------------------------------------------- + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct User { + id: u64, + name: String, + email: String, +} + +#[derive(Deserialize)] +struct NewUser { + name: String, + email: String, +} + +// --------------------------------------------------------------------------- +// Store actor protocol +// --------------------------------------------------------------------------- +// +// One enum per request kind. Each carries a `reply` Sender for the response. +// Reply types are kept simple: most are JSON byte vectors + HTTP status. The +// store does the serialisation; the handler just writes the bytes. + +enum Request { + List { reply: Sender<(u16, Vec)> }, + Get { id: u64, reply: Sender<(u16, Vec)> }, + Create { body: Vec, reply: Sender<(u16, Vec)> }, + Update { id: u64, body: Vec, reply: Sender<(u16, Vec)> }, + Delete { id: u64, reply: Sender<(u16, Vec)> }, +} + +const DB_PATH: &str = "/tmp/urus-crud.json"; + +// --------------------------------------------------------------------------- +// Store actor body +// --------------------------------------------------------------------------- + +fn store_loop(rx: smarm::Receiver) { + // Load on start. Missing file = empty store. Corrupt file = panic; we + // don't auto-rebuild because silently losing data is worse than failing + // loud. + let mut users: Vec = match std::fs::read(DB_PATH) { + Ok(bytes) if !bytes.is_empty() => + serde_json::from_slice(&bytes).expect("urus-crud: db file is not valid JSON"), + _ => Vec::new(), + }; + 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, + Err(_) => return, // all senders dropped + }; + match req { + Request::List { reply } => { + let body = serde_json::to_vec(&users).unwrap(); + let _ = reply.send((200, body)); + } + Request::Get { id, reply } => { + match users.iter().find(|u| u.id == id) { + Some(u) => { + let body = serde_json::to_vec(u).unwrap(); + let _ = reply.send((200, body)); + } + None => { + let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); + } + } + } + Request::Create { body, reply } => { + match serde_json::from_slice::(&body) { + Ok(nu) => { + let u = User { id: next_id, name: nu.name, email: nu.email }; + next_id += 1; + users.push(u.clone()); + persist(&users); + let _ = reply.send((201, serde_json::to_vec(&u).unwrap())); + } + Err(_) => { + let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec())); + } + } + } + Request::Update { id, body, reply } => { + match serde_json::from_slice::(&body) { + Ok(nu) => match users.iter_mut().find(|u| u.id == id) { + Some(u) => { + u.name = nu.name; + u.email = nu.email; + let snapshot = u.clone(); + persist(&users); + let _ = reply.send((200, serde_json::to_vec(&snapshot).unwrap())); + } + None => { + let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); + } + }, + Err(_) => { + let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec())); + } + } + } + Request::Delete { id, reply } => { + let before = users.len(); + users.retain(|u| u.id != id); + if users.len() < before { + persist(&users); + let _ = reply.send((204, Vec::new())); + } else { + let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); + } + } + } + } +} + +fn persist(users: &[User]) { + let json = serde_json::to_vec_pretty(users).unwrap(); + // Atomic-ish: write to temp then rename. Avoids half-written files on + // crash. /tmp is on the same filesystem so rename is atomic. + let tmp = format!("{DB_PATH}.tmp"); + std::fs::write(&tmp, &json).expect("urus-crud: write tmp failed"); + std::fs::rename(&tmp, DB_PATH).expect("urus-crud: rename failed"); +} + +// --------------------------------------------------------------------------- +// Handler helpers +// --------------------------------------------------------------------------- +// +// Once-cell trick: the store actor is spawned the first time a handler +// runs (smarm requires `spawn` to be called from inside an actor — which +// connection actors are). After that all handlers share the same Sender. +// Simpler than threading the Sender through the pipeline at startup. + +static STORE_TX: OnceLock> = OnceLock::new(); + +fn store() -> &'static Sender { + STORE_TX.get_or_init(|| { + let (tx, rx) = channel::(); + smarm::spawn(move || store_loop(rx)); + tx + }) +} + +fn json(conn: Conn, status: u16, body: Vec) -> Conn { + conn.put_status(status) + .put_header("content-type", "application/json") + .put_body(body) +} + +fn parse_id(s: &str) -> Option { + s.parse().ok() +} + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +fn list(conn: Conn, _next: Next) -> Conn { + let (tx, rx) = channel::<(u16, Vec)>(); + store().send(Request::List { reply: tx }).ok(); + let (status, body) = rx.recv().expect("store dropped"); + json(conn, status, body) +} + +fn create(conn: Conn, _next: Next) -> Conn { + let body = conn.body.as_bytes().to_vec(); + let (tx, rx) = channel::<(u16, Vec)>(); + store().send(Request::Create { body, reply: tx }).ok(); + let (status, body) = rx.recv().expect("store dropped"); + json(conn, status, body) +} + +fn get_one(conn: Conn, _next: Next) -> Conn { + let id = match conn.params.get("id").and_then(parse_id) { + Some(id) => id, + None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()), + }; + let (tx, rx) = channel::<(u16, Vec)>(); + store().send(Request::Get { id, reply: tx }).ok(); + let (status, body) = rx.recv().expect("store dropped"); + json(conn, status, body) +} + +fn update(conn: Conn, _next: Next) -> Conn { + let id = match conn.params.get("id").and_then(parse_id) { + Some(id) => id, + None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()), + }; + let body = conn.body.as_bytes().to_vec(); + let (tx, rx) = channel::<(u16, Vec)>(); + store().send(Request::Update { id, body, reply: tx }).ok(); + let (status, body) = rx.recv().expect("store dropped"); + json(conn, status, body) +} + +fn delete(conn: Conn, _next: Next) -> Conn { + let id = match conn.params.get("id").and_then(parse_id) { + Some(id) => id, + None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()), + }; + let (tx, rx) = channel::<(u16, Vec)>(); + store().send(Request::Delete { id, reply: tx }).ok(); + let (status, body) = rx.recv().expect("store dropped"); + json(conn, status, body) +} + +// --------------------------------------------------------------------------- +// Minimal request logger +// --------------------------------------------------------------------------- + +fn logger(conn: Conn, next: Next) -> Conn { + let method = conn.method.as_str().to_string(); + let path = conn.path.clone(); + let conn = next.run(conn); + println!("{method} {path} -> {}", conn.status.unwrap_or(0)); + conn +} + +// --------------------------------------------------------------------------- +// main +// --------------------------------------------------------------------------- + +fn main() { + let pipeline = Pipeline::new() + .plug(logger) + .plug( + Router::new() + .get( "/users", list) + .post( "/users", create) + .get( "/users/:id", get_one) + .put( "/users/:id", update) + .delete("/users/:id", delete), + ); + + let cfg = Config::new("127.0.0.1:8080".parse().unwrap()); + println!("urus-crud: DB at {DB_PATH}"); + serve_with(cfg, pipeline).unwrap(); +} diff --git a/src/conn.rs b/src/conn.rs new file mode 100644 index 0000000..cecb5b7 --- /dev/null +++ b/src/conn.rs @@ -0,0 +1,328 @@ +//! `Conn` — the value that flows through the entire plug pipeline. +//! +//! Carries request data, response state being built, and arbitrary user +//! data. Owned and moved at each step. Plugs are unaware of the underlying +//! HTTP version (v1.1 today, /2 later). + +use std::any::{Any, TypeId}; +use std::collections::HashMap; + +// --------------------------------------------------------------------------- +// Method +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum Method { + Get, + Post, + Put, + Delete, + Patch, + Head, + Options, + Other(String), +} + +impl Method { + pub fn parse(s: &str) -> Self { + match s { + "GET" => Method::Get, + "POST" => Method::Post, + "PUT" => Method::Put, + "DELETE" => Method::Delete, + "PATCH" => Method::Patch, + "HEAD" => Method::Head, + "OPTIONS" => Method::Options, + _ => Method::Other(s.to_string()), + } + } + + pub fn as_str(&self) -> &str { + match self { + Method::Get => "GET", + Method::Post => "POST", + Method::Put => "PUT", + Method::Delete => "DELETE", + Method::Patch => "PATCH", + Method::Head => "HEAD", + Method::Options => "OPTIONS", + Method::Other(s) => s.as_str(), + } + } +} + +// --------------------------------------------------------------------------- +// HttpVersion +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HttpVersion { + Http10, + Http11, +} + +impl HttpVersion { + pub fn as_str(self) -> &'static str { + match self { + HttpVersion::Http10 => "HTTP/1.0", + HttpVersion::Http11 => "HTTP/1.1", + } + } +} + +// --------------------------------------------------------------------------- +// HeaderMap — small-vec of (name, value) string pairs. +// --------------------------------------------------------------------------- +// +// Per spec: most requests have <20 headers, so a linear-scan Vec is cheaper +// than a hash map. Names are case-insensitive (HTTP/1.1), normalised to +// lowercase on insert. + +#[derive(Debug, Clone, Default)] +pub struct HeaderMap { + inner: Vec<(String, String)>, +} + +impl HeaderMap { + pub fn new() -> Self { Self::default() } + + pub fn with_capacity(n: usize) -> Self { + Self { inner: Vec::with_capacity(n) } + } + + /// Append a header. Does not deduplicate (HTTP allows repeats; e.g. + /// `Set-Cookie`). + pub fn append(&mut self, name: &str, value: impl Into) { + self.inner.push((name.to_ascii_lowercase(), value.into())); + } + + /// Replace any existing values for `name` with a single value. + pub fn set(&mut self, name: &str, value: impl Into) { + let lower = name.to_ascii_lowercase(); + self.inner.retain(|(n, _)| n != &lower); + self.inner.push((lower, value.into())); + } + + /// First value for `name`, if any. + pub fn get(&self, name: &str) -> Option<&str> { + let lower = name.to_ascii_lowercase(); + self.inner.iter() + .find(|(n, _)| n == &lower) + .map(|(_, v)| v.as_str()) + } + + /// Iterate over (name, value) pairs in insertion order. + pub fn iter(&self) -> impl Iterator { + self.inner.iter().map(|(n, v)| (n.as_str(), v.as_str())) + } + + pub fn len(&self) -> usize { self.inner.len() } + pub fn is_empty(&self) -> bool { self.inner.is_empty() } +} + +// --------------------------------------------------------------------------- +// Body — request body, exposed as owned bytes. +// --------------------------------------------------------------------------- +// +// v1: bodies are read in full by the connection actor before the pipeline +// runs, when `Content-Length` is set. Streaming and chunked encoding will +// come later; for now this is the simplest correct thing. + +#[derive(Debug, Clone, Default)] +pub struct Body { + bytes: Vec, +} + +impl Body { + pub fn empty() -> Self { Self::default() } + pub fn from_bytes(b: Vec) -> Self { Self { bytes: b } } + pub fn as_bytes(&self) -> &[u8] { &self.bytes } + pub fn into_bytes(self) -> Vec { self.bytes } + pub fn len(&self) -> usize { self.bytes.len() } + pub fn is_empty(&self) -> bool { self.bytes.is_empty() } +} + +// --------------------------------------------------------------------------- +// 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. + +#[derive(Debug, Clone)] +pub enum RespBody { + Empty, + Bytes(Vec), +} + +impl RespBody { + pub fn len_hint(&self) -> usize { + match self { + RespBody::Empty => 0, + RespBody::Bytes(b) => b.len(), + } + } +} + +impl Default for RespBody { + fn default() -> Self { RespBody::Empty } +} + +// --------------------------------------------------------------------------- +// Params — path parameters extracted by the router. +// --------------------------------------------------------------------------- +// +// Small-vec semantics: routes have a handful of params; a Vec is the right +// shape. Empty until the router plug populates it. + +#[derive(Debug, Clone, Default)] +pub struct Params { + inner: Vec<(String, String)>, +} + +impl Params { + pub fn new() -> Self { Self::default() } + + pub fn put(&mut self, name: impl Into, value: impl Into) { + self.inner.push((name.into(), value.into())); + } + + pub fn get(&self, name: &str) -> Option<&str> { + self.inner.iter() + .find(|(n, _)| n == name) + .map(|(_, v)| v.as_str()) + } + + pub fn iter(&self) -> impl Iterator { + self.inner.iter().map(|(n, v)| (n.as_str(), v.as_str())) + } +} + +// --------------------------------------------------------------------------- +// Assigns — type-erased map for arbitrary user data. +// --------------------------------------------------------------------------- +// +// Same pattern as Phoenix's `conn.assigns`. The `Option` keeps the +// common case (no assigns) at zero allocation cost. + +#[derive(Debug, Default)] +pub struct Assigns { + inner: Option>>, +} + +impl Assigns { + pub fn new() -> Self { Self::default() } + + pub fn put(&mut self, value: T) { + self.inner + .get_or_insert_with(HashMap::new) + .insert(TypeId::of::(), Box::new(value)); + } + + pub fn get(&self) -> Option<&T> { + self.inner + .as_ref()? + .get(&TypeId::of::())? + .downcast_ref::() + } + + pub fn take(&mut self) -> Option { + let boxed = self.inner.as_mut()?.remove(&TypeId::of::())?; + boxed.downcast::().ok().map(|b| *b) + } +} + +// --------------------------------------------------------------------------- +// Conn — the pipeline value. +// --------------------------------------------------------------------------- + +#[derive(Debug)] +pub struct Conn { + // Request (populated before the pipeline runs). + pub method: Method, + pub path: String, + pub query: Option, + pub version: HttpVersion, + pub headers: HeaderMap, + pub body: Body, + + // Response (built up by plugs, written after the pipeline returns). + pub status: Option, + pub resp_headers: HeaderMap, + pub resp_body: RespBody, + + // Pipeline state. + pub params: Params, + pub assigns: Assigns, + pub halted: bool, +} + +impl Conn { + /// Construct an empty `Conn`. Tests use this; the connection actor uses + /// `From` (see parser.rs). + pub fn new() -> Self { + Self { + method: Method::Get, + path: String::new(), + query: None, + version: HttpVersion::Http11, + headers: HeaderMap::new(), + body: Body::empty(), + + status: None, + resp_headers: HeaderMap::new(), + resp_body: RespBody::Empty, + + params: Params::new(), + assigns: Assigns::new(), + halted: false, + } + } + + // ----- Fluent builders. Most handler code uses these. ----- + + pub fn put_status(mut self, status: u16) -> Self { + self.status = Some(status); + self + } + + pub fn put_header(mut self, name: &str, value: impl Into) -> Self { + self.resp_headers.set(name, value); + self + } + + pub fn put_body(mut self, body: impl Into) -> Self { + self.resp_body = body.into(); + self + } + + pub fn put_params(mut self, params: Params) -> Self { + self.params = params; + self + } + + pub fn halt(mut self) -> Self { + self.halted = true; + self + } +} + +impl Default for Conn { + fn default() -> Self { Self::new() } +} + +// ----- Ergonomic conversions into RespBody. ----- + +impl From> for RespBody { + fn from(v: Vec) -> Self { RespBody::Bytes(v) } +} +impl From for RespBody { + fn from(s: String) -> Self { RespBody::Bytes(s.into_bytes()) } +} +impl From<&'static str> for RespBody { + fn from(s: &'static str) -> Self { RespBody::Bytes(s.as_bytes().to_vec()) } +} +impl From<&[u8]> for RespBody { + fn from(s: &[u8]) -> Self { RespBody::Bytes(s.to_vec()) } +} diff --git a/src/conn_actor.rs b/src/conn_actor.rs new file mode 100644 index 0000000..6ca80c5 --- /dev/null +++ b/src/conn_actor.rs @@ -0,0 +1,309 @@ +//! The connection actor — one per accepted TCP connection. +//! +//! Runs the HTTP/1.1 request loop: +//! +//! loop { +//! read bytes → parse → build Conn +//! pipeline.run(conn) // inline; no spawn +//! write response +//! if !keep_alive { break } +//! } +//! +//! Everything in here happens in one smarm green thread. The actor parks on +//! `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::net::OwnedFd; +use crate::parser::{self, ParseError}; +use crate::plug::Pipeline; + +use std::io::{self, ErrorKind}; +use std::os::fd::RawFd; +use std::time::Duration; + +// --------------------------------------------------------------------------- +// Limits +// --------------------------------------------------------------------------- + +/// Per-connection settings the connection actor needs to honour. +#[derive(Clone, Copy, Debug)] +pub struct ConnLimits { + pub max_headers: usize, + pub initial_read_buf: usize, + /// Hard cap on the request head to bound buffer growth. 64 KiB is + /// well over Apache's 8 KiB default; protects against pathological + /// clients streaming headers forever. + pub max_head_bytes: usize, + /// Hard cap on Content-Length we'll accept. 16 MiB is enough for a CRUD + /// example; configurable in `Config`. + pub max_body_bytes: usize, + pub keep_alive_timeout: Duration, +} + +impl Default for ConnLimits { + fn default() -> Self { + Self { + max_headers: 64, + initial_read_buf: 8 * 1024, + max_head_bytes: 64 * 1024, + max_body_bytes: 16 * 1024 * 1024, + keep_alive_timeout: Duration::from_secs(60), + } + } +} + +// --------------------------------------------------------------------------- +// run_connection — entry point spawned by the listener actor. +// --------------------------------------------------------------------------- + +pub fn run_connection(fd: OwnedFd, pipeline: Pipeline, limits: ConnLimits) { + // The OwnedFd cleans up via Drop on any exit path (panic, error, or + // normal close). No explicit close calls below. + let raw = fd.as_raw(); + let mut buf: Vec = Vec::with_capacity(limits.initial_read_buf); + + loop { + // ----- 1. Read until we have a full request head. ----- + let parsed = 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::Io(_)) => { + // Network error or timeout. Best-effort close; we're done. + return; + } + Err(ReadHeadErr::Parse(e)) => { + emit_error_response(raw, &e); + return; + } + }; + + // ----- 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"); + return; + } + + // If client sent `Expect: 100-continue`, emit it before reading the + // 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() { + return; + } + } + + let body = match read_body(raw, &mut buf, parsed.head_len, body_len) { + Ok(b) => b, + Err(_) => return, + }; + + let keep_alive = parsed.keep_alive; + let version = parsed.version; + let head_len = parsed.head_len; + let conn = parser::build_conn(parsed, Body::from_bytes(body)); + + // ----- 3. Run the pipeline. ----- + // Catch panics at the actor boundary — a panicking handler should + // not take down the whole connection silently with no response. + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + pipeline.run(conn) + })); + + let mut response_conn = match result { + Ok(c) => c, + Err(_) => { + // Compose a 500 manually; the original Conn was moved into + // the closure. + let mut c = Conn::new(); + c.version = version; + c.put_status(500).put_header("content-length", "0") + } + }; + + // If no plug touched status, that's a configuration error (no router + // matched, no default handler). Emit 404. + if response_conn.status.is_none() { + response_conn = response_conn.put_status(404) + .put_body(RespBody::Empty); + } + + // ----- 4. Write the response. ----- + let bytes = parser::serialise_response(&response_conn, keep_alive); + if write_all(raw, &bytes).is_err() { + return; + } + + // ----- 5. Loop or close. ----- + if !keep_alive { + return; + } + + // Drop the request bytes (head + body) from `buf`; anything past + // them is the start of the next pipelined request. + let consumed = head_len + body_len; + buf.drain(..consumed); + } +} + +// --------------------------------------------------------------------------- +// read_head +// --------------------------------------------------------------------------- + +#[allow(dead_code)] // io::Error is captured for future logging +enum ReadHeadErr { + ClientClosed, + Io(io::Error), + Parse(ParseError), +} + +/// 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. +fn read_head( + fd: RawFd, + buf: &mut Vec, + limits: &ConnLimits, +) -> Result { + 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), + Err(ParseError::Incomplete) => {} // need more bytes + Err(e) => return Err(ReadHeadErr::Parse(e)), + } + } + + if buf.len() >= limits.max_head_bytes { + return Err(ReadHeadErr::Parse(ParseError::TooManyHeaders)); + } + + // Read more. + match read_some(fd, buf, limits.initial_read_buf) { + Ok(0) => return Err(ReadHeadErr::ClientClosed), + Ok(_) => continue, + Err(e) => return Err(ReadHeadErr::Io(e)), + } + } +} + +// --------------------------------------------------------------------------- +// read_body +// --------------------------------------------------------------------------- + +fn read_body( + fd: RawFd, + buf: &mut Vec, + head_len: usize, + body_len: usize, +) -> io::Result> { + // Bytes already in `buf` past the head belong to the body. + let already = buf.len().saturating_sub(head_len); + let need = body_len.saturating_sub(already); + + if need == 0 { + // We have the full body in `buf` already. Extract a copy; `buf` is + // drained later in the connection loop. + return Ok(buf[head_len..head_len + body_len].to_vec()); + } + + // Read until we have the rest. + let mut total_read = already; + while total_read < body_len { + match read_some(fd, buf, 8 * 1024) { + Ok(0) => return Err(io::Error::new(ErrorKind::UnexpectedEof, "client closed during body")), + Ok(n) => total_read += n, + Err(e) => return Err(e), + } + } + Ok(buf[head_len..head_len + body_len].to_vec()) +} + +// --------------------------------------------------------------------------- +// read_some — single epoll-park + read loop. +// --------------------------------------------------------------------------- +// +// Appends what it reads onto `buf`. Returns bytes read, 0 for EOF, or the +// last io error. + +fn read_some(fd: RawFd, buf: &mut Vec, chunk: usize) -> 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). + loop { + smarm::wait_readable(fd)?; + + let start = buf.len(); + buf.resize(start + chunk, 0); + + let n = unsafe { + libc::read(fd, buf.as_mut_ptr().add(start) as *mut _, chunk) + }; + + if n < 0 { + let err = io::Error::last_os_error(); + buf.truncate(start); + if err.kind() == ErrorKind::WouldBlock || err.kind() == ErrorKind::Interrupted { + continue; + } + return Err(err); + } + + let n = n as usize; + buf.truncate(start + n); + return Ok(n); // n == 0 here is real EOF + } +} + +// --------------------------------------------------------------------------- +// write_all — robust write loop. +// --------------------------------------------------------------------------- + +fn write_all(fd: RawFd, mut buf: &[u8]) -> io::Result<()> { + while !buf.is_empty() { + // Park on writability before each syscall. + smarm::wait_writable(fd)?; + + let n = unsafe { + libc::write(fd, buf.as_ptr() as *const _, buf.len()) + }; + if n < 0 { + let err = io::Error::last_os_error(); + if err.kind() == ErrorKind::WouldBlock { + continue; // spurious wake; retry + } + return Err(err); + } + if n == 0 { + return Err(io::Error::new(ErrorKind::WriteZero, "write returned 0")); + } + buf = &buf[n as usize..]; + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Error responses for unparseable / malformed requests. +// --------------------------------------------------------------------------- + +fn emit_error_response(fd: RawFd, err: &ParseError) { + 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", + ParseError::BadContentLength => + b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + ParseError::Unsupported => + b"HTTP/1.1 411 Length Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + // Incomplete and Malformed both lead here; Incomplete shouldn't + // appear (read_head loops on it). + _ => + b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + }; + let _ = write_all(fd, resp); +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..bbcc999 --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,33 @@ +//! # urus — a cowboy/bandit-style HTTP library for the smarm actor runtime. +//! +//! v1 covers HTTP/1.1, the plug pipeline, and a built-in router. See +//! `urus-spec.md` for the design. +//! +//! ```no_run +//! use urus::{Pipeline, Router, Conn, Next, serve}; +//! +//! let pipeline = Pipeline::new().plug( +//! Router::new() +//! .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello")) +//! .get("/users/:id", |c: Conn, _n: Next| { +//! let id = c.params.get("id").unwrap_or("").to_string(); +//! c.put_status(200).put_body(id) +//! }) +//! ); +//! +//! serve("0.0.0.0:8080", pipeline).unwrap(); +//! ``` + +pub mod conn; +pub mod plug; +pub mod router; +pub mod parser; +pub mod net; +pub mod conn_actor; +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 plug::{Next, Pipeline, Plug}; +pub use router::Router; +pub use serve::{serve, serve_with, Config}; diff --git a/src/net.rs b/src/net.rs new file mode 100644 index 0000000..701711c --- /dev/null +++ b/src/net.rs @@ -0,0 +1,165 @@ +//! Thin TCP socket layer. +//! +//! We don't use `std::net::TcpListener` because we need non-blocking accept +//! integrated with smarm's epoll loop. So this drops down to libc: socket, +//! bind, listen, accept4. All fds carry `O_NONBLOCK | O_CLOEXEC` so smarm's +//! readiness primitives work as documented. + +use std::io; +use std::net::SocketAddr; +use std::os::fd::RawFd; + +// --------------------------------------------------------------------------- +// OwnedFd — a tiny RAII wrapper that closes on drop. +// --------------------------------------------------------------------------- +// +// Sent through smarm channels (which require `Send`) for zero-copy fd +// handoff between listener and connection actor. `Send` is safe: fd values +// are just integers, and ownership semantics — exactly-one closer — are +// enforced by the type itself. + +#[derive(Debug)] +pub struct OwnedFd { + fd: RawFd, +} + +impl OwnedFd { + /// Wrap a raw fd. The wrapper now owns the fd and will close it on drop. + pub fn from_raw(fd: RawFd) -> Self { + Self { fd } + } + + pub fn as_raw(&self) -> RawFd { self.fd } + + /// Release ownership without closing. The caller must close the fd. + pub fn into_raw(self) -> RawFd { + let fd = self.fd; + std::mem::forget(self); + fd + } +} + +impl Drop for OwnedFd { + fn drop(&mut self) { + if self.fd >= 0 { + unsafe { libc::close(self.fd); } + } + } +} + +// fd handoff: SAFETY: RawFd is an integer; sending one across threads +// transfers ownership in the same way moving an i32 would. The recipient +// becomes the unique closer. +unsafe impl Send for OwnedFd {} + +// --------------------------------------------------------------------------- +// bind_and_listen +// --------------------------------------------------------------------------- + +const LISTEN_BACKLOG: i32 = 1024; + +/// Create a non-blocking, SO_REUSEADDR TCP listener bound to `addr`. The +/// returned fd is ready for `accept4` calls; the caller registers it with +/// smarm via `wait_readable` between accepts. +pub fn bind_and_listen(addr: SocketAddr) -> io::Result { + let family = match addr { + SocketAddr::V4(_) => libc::AF_INET, + SocketAddr::V6(_) => libc::AF_INET6, + }; + + let fd = unsafe { + libc::socket( + family, + libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + ) + }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + let owned = OwnedFd::from_raw(fd); + + // SO_REUSEADDR — standard for servers; avoids TIME_WAIT bind failures + // on restart. + let opt: libc::c_int = 1; + let r = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_REUSEADDR, + &opt as *const _ as *const libc::c_void, + std::mem::size_of_val(&opt) as libc::socklen_t, + ) + }; + if r < 0 { + return Err(io::Error::last_os_error()); + } + + // bind(). + match addr { + SocketAddr::V4(a) => { + let sa = libc::sockaddr_in { + sin_family: libc::AF_INET as u16, + sin_port: a.port().to_be(), + sin_addr: libc::in_addr { s_addr: u32::from_ne_bytes(a.ip().octets()) }, + sin_zero: [0; 8], + }; + let r = unsafe { + libc::bind( + fd, + &sa as *const _ as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r < 0 { return Err(io::Error::last_os_error()); } + } + SocketAddr::V6(a) => { + let sa = libc::sockaddr_in6 { + sin6_family: libc::AF_INET6 as u16, + sin6_port: a.port().to_be(), + sin6_flowinfo: a.flowinfo(), + sin6_addr: libc::in6_addr { s6_addr: a.ip().octets() }, + sin6_scope_id: a.scope_id(), + }; + let r = unsafe { + libc::bind( + fd, + &sa as *const _ as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if r < 0 { return Err(io::Error::last_os_error()); } + } + } + + // listen(). + let r = unsafe { libc::listen(fd, LISTEN_BACKLOG) }; + if r < 0 { return Err(io::Error::last_os_error()); } + + Ok(owned) +} + +// --------------------------------------------------------------------------- +// accept_nonblocking +// --------------------------------------------------------------------------- + +/// One non-blocking `accept4`. Returns the new fd on success, +/// `Err(WouldBlock)` if no connection is pending (caller should park on +/// `wait_readable(listener)` and retry), or other errors directly. +pub fn accept_nonblocking(listener: RawFd) -> io::Result { + let mut addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut len: libc::socklen_t = std::mem::size_of::() as libc::socklen_t; + + let fd = unsafe { + libc::accept4( + listener, + &mut addr as *mut _ as *mut libc::sockaddr, + &mut len, + libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + ) + }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + Ok(OwnedFd::from_raw(fd)) +} diff --git a/src/parser.rs b/src/parser.rs new file mode 100644 index 0000000..289e18f --- /dev/null +++ b/src/parser.rs @@ -0,0 +1,379 @@ +//! HTTP/1.1 wire protocol — request parsing and response serialisation. +//! +//! Parsing uses `httparse` for the request line + headers (zero-alloc, zero- +//! copy, battle-tested). Body framing, keep-alive logic and response writing +//! are ours. +//! +//! v1 body framing: +//! - `Content-Length: N` — read exactly N bytes. +//! - No body header — empty body. +//! - `Transfer-Encoding: chunked` — deferred. Returns ParseError::Unsupported +//! and the connection actor responds 411 Length Required + close. +//! +//! Keep it stupid simple. Chunked decoding lands when something actually +//! requests it. + +use crate::conn::{Body, Conn, HeaderMap, HttpVersion, Method, RespBody}; + +// --------------------------------------------------------------------------- +// ParseError +// --------------------------------------------------------------------------- + +#[derive(Debug)] +pub enum ParseError { + /// Request bytes are well-formed but incomplete; need more from the + /// socket. + Incomplete, + /// Request is malformed; respond 400 and close. + Malformed, + /// Header count exceeded the configured max; respond 431 and close. + TooManyHeaders, + /// `Content-Length` header could not be parsed as an integer. + BadContentLength, + /// A wire feature we haven't implemented yet (e.g. chunked encoding). + /// Connection actor responds 411 + close. + Unsupported, +} + +// --------------------------------------------------------------------------- +// ParsedHead — what `parse_head` returns on success. +// --------------------------------------------------------------------------- +// +// `head_len` is how many bytes from the start of the buffer the head +// occupied — the body starts at `&buf[head_len..]`. + +pub struct ParsedHead { + pub head_len: usize, + pub method: Method, + pub path: String, + pub query: Option, + pub version: HttpVersion, + pub headers: HeaderMap, + pub content_length: Option, + pub keep_alive: bool, + pub expect_100: bool, +} + +/// Try to parse a request head from `buf`. Returns `Incomplete` if more +/// bytes are needed; the caller should read more and retry with the *same* +/// buffer. +pub fn parse_head(buf: &[u8], max_headers: usize) -> Result { + // httparse needs a header array; size it from config. 64 is the spec + // default. Stack allocation; if a request has more than max_headers + // they get TooManyHeaders. + let mut header_buf = vec![httparse::EMPTY_HEADER; max_headers]; + let mut req = httparse::Request::new(&mut header_buf); + + let head_len = match req.parse(buf) { + Ok(httparse::Status::Complete(n)) => n, + Ok(httparse::Status::Partial) => return Err(ParseError::Incomplete), + Err(httparse::Error::TooManyHeaders) => return Err(ParseError::TooManyHeaders), + Err(_) => return Err(ParseError::Malformed), + }; + + let method_str = req.method.ok_or(ParseError::Malformed)?; + let raw_path = req.path.ok_or(ParseError::Malformed)?; + let v = req.version.ok_or(ParseError::Malformed)?; + + let version = match v { + 0 => HttpVersion::Http10, + 1 => HttpVersion::Http11, + _ => return Err(ParseError::Malformed), + }; + + // Split path and query. We don't percent-decode the path here — the + // spec calls for it, but for v1 raw is fine; user code can decode if + // they need to. Same with the query. + let (path, query) = match raw_path.split_once('?') { + Some((p, q)) => (p.to_string(), Some(q.to_string())), + None => (raw_path.to_string(), None), + }; + + // Walk headers, building HeaderMap and pulling out the few we need + // ourselves (Content-Length, Connection, Transfer-Encoding, Expect). + let mut headers = HeaderMap::with_capacity(req.headers.len()); + let mut content_length = None; + let mut connection_hdr = None; + let mut chunked = false; + let mut expect_100 = false; + + for h in req.headers.iter() { + let name_lower = h.name.to_ascii_lowercase(); + let value = std::str::from_utf8(h.value).map_err(|_| ParseError::Malformed)?; + + match name_lower.as_str() { + "content-length" => { + content_length = Some( + value.trim() + .parse::() + .map_err(|_| ParseError::BadContentLength)? + ); + } + "transfer-encoding" => { + // We only care whether it includes "chunked". Multiple codings + // can appear; chunked is the only one we'd need to decode. + if value.to_ascii_lowercase().split(',').any(|t| t.trim() == "chunked") { + chunked = true; + } + } + "connection" => { + connection_hdr = Some(value.to_ascii_lowercase()); + } + "expect" => { + if value.eq_ignore_ascii_case("100-continue") { + expect_100 = true; + } + } + _ => {} + } + headers.append(&name_lower, value.to_string()); + } + + if chunked { + return Err(ParseError::Unsupported); + } + + // Keep-alive logic, RFC 7230 §6.3: + // HTTP/1.1: keep-alive by default; "Connection: close" overrides. + // HTTP/1.0: close by default; "Connection: keep-alive" overrides. + let keep_alive = match version { + HttpVersion::Http11 => connection_hdr.as_deref() != Some("close"), + HttpVersion::Http10 => connection_hdr.as_deref() == Some("keep-alive"), + }; + + Ok(ParsedHead { + head_len, + method: Method::parse(method_str), + path, + query, + version, + headers, + content_length, + keep_alive, + expect_100, + }) +} + +// --------------------------------------------------------------------------- +// Conn assembly +// --------------------------------------------------------------------------- + +pub fn build_conn(head: ParsedHead, body: Body) -> Conn { + let mut c = Conn::new(); + c.method = head.method; + c.path = head.path; + c.query = head.query; + c.version = head.version; + c.headers = head.headers; + c.body = body; + c +} + +// --------------------------------------------------------------------------- +// Response serialisation +// --------------------------------------------------------------------------- +// +// We emit: +// \r\n +//

: \r\n +// ... +// \r\n +// +// +// Headers we always inject (unless the user already set them): +// - Content-Length: from resp_body.len_hint() +// - Connection: close (if keep-alive is off this request) +// - Date: skipped in v1; not required by the standard and adds complexity. + +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 status = conn.status.unwrap_or(200); + let reason = reason_phrase(status); + + // Status line. + out.extend_from_slice(conn.version.as_str().as_bytes()); + out.push(b' '); + out.extend_from_slice(status.to_string().as_bytes()); + out.push(b' '); + out.extend_from_slice(reason.as_bytes()); + out.extend_from_slice(b"\r\n"); + + // User headers — written first so subsequent injection can skip them. + let mut wrote_content_length = false; + let mut wrote_connection = false; + + for (name, value) in conn.resp_headers.iter() { + match name { + "content-length" => wrote_content_length = true, + "connection" => wrote_connection = true, + _ => {} + } + out.extend_from_slice(name.as_bytes()); + out.extend_from_slice(b": "); + out.extend_from_slice(value.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + + 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"); + } + + if !wrote_connection && !keep_alive { + out.extend_from_slice(b"connection: close\r\n"); + } + + out.extend_from_slice(b"\r\n"); + + // Body. + match &conn.resp_body { + RespBody::Empty => {} + RespBody::Bytes(b) => out.extend_from_slice(b), + } + + out +} + +/// HTTP/1.1 reason phrases. Required by RFC 7230 §3.1.2 — clients may +/// display them. +pub fn reason_phrase(status: u16) -> &'static str { + match status { + 200 => "OK", + 201 => "Created", + 202 => "Accepted", + 204 => "No Content", + 301 => "Moved Permanently", + 302 => "Found", + 303 => "See Other", + 304 => "Not Modified", + 307 => "Temporary Redirect", + 308 => "Permanent Redirect", + 400 => "Bad Request", + 401 => "Unauthorized", + 403 => "Forbidden", + 404 => "Not Found", + 405 => "Method Not Allowed", + 409 => "Conflict", + 411 => "Length Required", + 413 => "Payload Too Large", + 414 => "URI Too Long", + 415 => "Unsupported Media Type", + 422 => "Unprocessable Entity", + 429 => "Too Many Requests", + 431 => "Request Header Fields Too Large", + 500 => "Internal Server Error", + 501 => "Not Implemented", + 502 => "Bad Gateway", + 503 => "Service Unavailable", + _ => "OK", // best-effort fallback; spec allows any phrase + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_simple_get() { + let req = b"GET /hello HTTP/1.1\r\nHost: x\r\n\r\n"; + let head = parse_head(req, 64).unwrap(); + assert_eq!(head.method, Method::Get); + assert_eq!(head.path, "/hello"); + assert_eq!(head.version, HttpVersion::Http11); + assert!(head.keep_alive); + assert_eq!(head.content_length, None); + } + + #[test] + fn parse_with_query() { + let req = b"GET /users?id=42&active=true HTTP/1.1\r\nHost: x\r\n\r\n"; + let head = parse_head(req, 64).unwrap(); + assert_eq!(head.path, "/users"); + assert_eq!(head.query.as_deref(), Some("id=42&active=true")); + } + + #[test] + fn parse_post_with_content_length() { + let req = b"POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\nhello"; + let head = parse_head(req, 64).unwrap(); + assert_eq!(head.content_length, Some(5)); + // head_len is bytes before the body; the remainder is the body. + assert_eq!(&req[head.head_len..], b"hello"); + } + + #[test] + fn parse_connection_close() { + let req = b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"; + let head = parse_head(req, 64).unwrap(); + assert!(!head.keep_alive); + } + + #[test] + fn parse_http10_default_close() { + let req = b"GET / HTTP/1.0\r\nHost: x\r\n\r\n"; + let head = parse_head(req, 64).unwrap(); + assert!(!head.keep_alive); + } + + #[test] + fn parse_http10_keepalive_opt_in() { + let req = b"GET / HTTP/1.0\r\nHost: x\r\nConnection: keep-alive\r\n\r\n"; + let head = parse_head(req, 64).unwrap(); + assert!(head.keep_alive); + } + + #[test] + fn parse_partial() { + let req = b"GET /hello HTTP/1.1\r\nHost: x\r\n"; // no terminator yet + match parse_head(req, 64) { + Err(ParseError::Incomplete) => {} + _ => panic!("expected Incomplete"), + } + } + + #[test] + fn parse_chunked_unsupported() { + let req = b"POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n"; + match parse_head(req, 64) { + Err(ParseError::Unsupported) => {} + _ => panic!("expected Unsupported for chunked"), + } + } + + #[test] + fn serialise_basic_200() { + let conn = Conn::new().put_status(200).put_body("hi"); + let bytes = serialise_response(&conn, true); + let s = std::str::from_utf8(&bytes).unwrap(); + assert!(s.starts_with("HTTP/1.1 200 OK\r\n")); + assert!(s.contains("content-length: 2")); + assert!(s.ends_with("\r\n\r\nhi")); + } + + #[test] + fn serialise_connection_close_when_not_keepalive() { + let conn = Conn::new().put_status(204); + let bytes = serialise_response(&conn, false); + let s = std::str::from_utf8(&bytes).unwrap(); + assert!(s.contains("connection: close")); + } + + #[test] + fn serialise_user_content_length_is_respected() { + let conn = Conn::new().put_status(200) + .put_header("content-length", "999") + .put_body("hi"); // mismatch on purpose — user wins + let bytes = serialise_response(&conn, true); + let s = std::str::from_utf8(&bytes).unwrap(); + assert!(s.contains("content-length: 999")); + assert!(!s.contains("content-length: 2")); + } +} diff --git a/src/plug.rs b/src/plug.rs new file mode 100644 index 0000000..25303bd --- /dev/null +++ b/src/plug.rs @@ -0,0 +1,100 @@ +//! The plug pipeline — universal abstraction for everything in urus. +//! +//! A `Plug` is anything that takes a `Conn`, optionally calls `next.run(conn)` +//! to delegate to the rest of the pipeline, and returns a `Conn`. Routing, +//! middleware, handlers — all the same shape. +//! +//! The halt is structural: not calling `next.run(conn)` *is* the halt. Setting +//! `conn.halted = true` is only a hint to outer plugs that the pipeline was +//! short-circuited (e.g. a logger that wraps `next`). + +use crate::conn::Conn; +use std::sync::Arc; + +// --------------------------------------------------------------------------- +// Plug trait +// --------------------------------------------------------------------------- + +pub trait Plug: Send + Sync + 'static { + fn call(&self, conn: Conn, next: Next) -> Conn; +} + +// Blanket impl for closures. Both `Send + Sync` so plug values can be shared +// across connection actors via the Arc'd pipeline. +impl Plug for F +where + F: Fn(Conn, Next) -> Conn + Send + Sync + 'static, +{ + fn call(&self, conn: Conn, next: Next) -> Conn { + self(conn, next) + } +} + +// --------------------------------------------------------------------------- +// Next — a handle to the remainder of the pipeline. +// --------------------------------------------------------------------------- +// +// Opaque wrapper around a slice of the remaining plugs. `run(conn)` consumes +// the handle (no double-calling) and dispatches to the head plug. Empty +// remainder is a no-op — the conn passes through unchanged. + +pub struct Next<'a> { + plugs: &'a [Arc], +} + +impl<'a> Next<'a> { + pub(crate) fn new(plugs: &'a [Arc]) -> Self { + Self { plugs } + } + + /// Continue down the pipeline. Returns the `Conn` produced by the rest. + pub fn run(self, conn: Conn) -> Conn { + match self.plugs.split_first() { + None => conn, + Some((head, tail)) => head.call(conn, Next { plugs: tail }), + } + } +} + +// --------------------------------------------------------------------------- +// Pipeline +// --------------------------------------------------------------------------- +// +// Built once at startup, cloned cheaply (Arc bump) into each connection +// actor. Read-only after construction; no per-request allocation in here. + +#[derive(Clone)] +pub struct Pipeline { + plugs: Arc>>, +} + +impl Pipeline { + pub fn new() -> Self { + Self { plugs: Arc::new(Vec::new()) } + } + + /// Append a plug. Builder pattern — consumes and returns self. + /// + /// Mutates the underlying Vec via `Arc::make_mut` when possible (it's + /// unique during construction), which is allocation-free after the first + /// `plug()` until clones are taken. + pub fn plug(mut self, p: impl Plug) -> Self { + let v = Arc::make_mut(&mut self.plugs); + v.push(Arc::new(p)); + self + } + + /// Run the pipeline against a `Conn`. The connection actor calls this + /// once per request. + pub fn run(&self, conn: Conn) -> Conn { + Next::new(self.plugs.as_slice()).run(conn) + } + + /// Number of plugs in the pipeline. Mostly for tests. + pub fn len(&self) -> usize { self.plugs.len() } + pub fn is_empty(&self) -> bool { self.plugs.is_empty() } +} + +impl Default for Pipeline { + fn default() -> Self { Self::new() } +} diff --git a/src/router.rs b/src/router.rs new file mode 100644 index 0000000..dd41dd5 --- /dev/null +++ b/src/router.rs @@ -0,0 +1,223 @@ +//! The router — itself a plug. +//! +//! v1: linear scan over a Vec of compiled route patterns. The spec calls for +//! a radix trie eventually; for "good enough to build a CRUD app" with +//! typical route counts (<50), the linear scan is dominant noise next to +//! everything else on the request path. Replace with a trie when there's +//! evidence it matters. +//! +//! Pattern syntax: +//! - `/users` — literal +//! - `/users/:id` — `id` param +//! - `/users/:id/posts/:n` — multiple params +//! +//! No wildcards in v1. They are not needed for CRUD. + +use crate::conn::{Conn, Method, Params}; +use crate::plug::{Next, Plug}; +use std::sync::Arc; + +// --------------------------------------------------------------------------- +// Pattern +// --------------------------------------------------------------------------- + +/// A compiled URL pattern: a list of segments, each either a literal or a +/// named parameter. Segment compilation happens at `Router::get(...)` time; +/// matching at request time is a single-pass loop. +#[derive(Debug, Clone)] +struct Pattern { + segments: Vec, +} + +#[derive(Debug, Clone)] +enum Segment { + Literal(String), + Param(String), +} + +impl Pattern { + fn compile(path: &str) -> Self { + // Treat empty path the same as "/" — single empty segment. + let trimmed = path.trim_start_matches('/'); + if trimmed.is_empty() { + return Pattern { segments: Vec::new() }; + } + let segments = trimmed + .split('/') + .map(|s| { + if let Some(name) = s.strip_prefix(':') { + Segment::Param(name.to_string()) + } else { + Segment::Literal(s.to_string()) + } + }) + .collect(); + Pattern { segments } + } + + /// Match `path` against this pattern. Returns extracted params on + /// success, `None` on miss. Strict path equality — trailing slashes + /// matter; that's standard HTTP behaviour and we don't paper over it. + fn match_path(&self, path: &str) -> Option { + let trimmed = path.trim_start_matches('/'); + if trimmed.is_empty() { + return if self.segments.is_empty() { Some(Params::new()) } else { None }; + } + let parts: Vec<&str> = trimmed.split('/').collect(); + if parts.len() != self.segments.len() { + return None; + } + let mut params = Params::new(); + for (seg, part) in self.segments.iter().zip(parts.iter()) { + match seg { + Segment::Literal(lit) => { + if lit != part { + return None; + } + } + Segment::Param(name) => { + params.put(name.clone(), (*part).to_string()); + } + } + } + Some(params) + } +} + +// --------------------------------------------------------------------------- +// Route +// --------------------------------------------------------------------------- + +struct Route { + method: Method, + pattern: Pattern, + handler: Arc, +} + +// --------------------------------------------------------------------------- +// Router +// --------------------------------------------------------------------------- + +pub struct Router { + routes: Vec, +} + +impl Router { + pub fn new() -> Self { + Self { routes: Vec::new() } + } + + fn add(mut self, method: Method, path: &str, plug: impl Plug) -> Self { + self.routes.push(Route { + method, + pattern: Pattern::compile(path), + handler: Arc::new(plug), + }); + self + } + + pub fn get(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Get, path, plug) } + pub fn post(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Post, path, plug) } + pub fn put(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Put, path, plug) } + pub fn delete(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Delete, path, plug) } + pub fn patch(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Patch, path, plug) } + pub fn head(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Head, path, plug) } + pub fn options(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Options, path, plug) } +} + +impl Default for Router { + fn default() -> Self { Self::new() } +} + +impl Plug for Router { + fn call(&self, conn: Conn, next: Next) -> Conn { + // Two-pass dispatch: first match on method+path; if no method-and- + // path matches but a same-path-different-method does, return 405. + // Otherwise pass through to `next` so outer pipelines can layer a + // 404 handler (or skip and let the connection actor emit a default). + let mut path_seen = false; + for route in &self.routes { + if let Some(params) = route.pattern.match_path(&conn.path) { + if route.method == conn.method { + let conn = conn.put_params(params); + return route.handler.call(conn, next); + } + path_seen = true; + } + } + if path_seen { + // Path is known, method isn't — RFC 7231 §6.5.5. + conn.put_status(405) + .put_header("content-length", "0") + .halt() + } else { + // No route matched at all; let outer plugs / default handler + // deal with it. + next.run(conn) + } + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::plug::Pipeline; + + fn ok_handler(_label: &'static str) -> impl Plug { + move |conn: Conn, _next: Next| conn.put_status(200).put_body(_label) + } + + #[test] + fn literal_match() { + let r = Router::new().get("/ping", ok_handler("pong")); + let mut c = Conn::new(); + c.method = Method::Get; + c.path = "/ping".into(); + let out = Pipeline::new().plug(r).run(c); + assert_eq!(out.status, Some(200)); + } + + #[test] + fn param_match() { + let r = Router::new().get("/users/:id", |conn: Conn, _n: Next| { + let id = conn.params.get("id").unwrap_or("").to_string(); + conn.put_status(200).put_body(id) + }); + let mut c = Conn::new(); + c.method = Method::Get; + c.path = "/users/42".into(); + let out = Pipeline::new().plug(r).run(c); + assert_eq!(out.status, Some(200)); + match &out.resp_body { + crate::conn::RespBody::Bytes(b) => assert_eq!(b, b"42"), + _ => panic!("expected bytes"), + } + } + + #[test] + fn method_mismatch_yields_405() { + let r = Router::new().get("/users", ok_handler("list")); + let mut c = Conn::new(); + c.method = Method::Post; + c.path = "/users".into(); + let out = Pipeline::new().plug(r).run(c); + assert_eq!(out.status, Some(405)); + } + + #[test] + fn no_route_falls_through() { + // Router returns conn unchanged; next plug (a 404 default) handles it. + let r = Router::new().get("/users", ok_handler("list")); + let default_404 = |conn: Conn, _n: Next| + conn.put_status(404).put_body("not found"); + let mut c = Conn::new(); + c.method = Method::Get; + c.path = "/nowhere".into(); + let out = Pipeline::new().plug(r).plug(default_404).run(c); + assert_eq!(out.status, Some(404)); + } +} diff --git a/src/serve.rs b/src/serve.rs new file mode 100644 index 0000000..6baf7af --- /dev/null +++ b/src/serve.rs @@ -0,0 +1,188 @@ +//! Listener pool and the `serve` entry point. +//! +//! A small fixed pool of listener actors share the same TCP listen fd (via +//! `dup`) — each blocks in non-blocking `accept4` + `wait_readable` on its +//! own copy. When a connection arrives the listener spawns a connection +//! actor with the `OwnedFd` and immediately returns to `accept`. No +//! coordination needed; the kernel serialises `accept` calls across the fds. +//! +//! Sharing via `dup` rather than the same fd is deliberate — Linux's +//! `accept4` is thread-safe on a single fd, but dup'ing per-listener keeps +//! each actor's epoll registration local to its own RawFd value (so smarm's +//! `waiters: HashMap` doesn't see collisions between listeners +//! waiting on "the same fd"). + +use crate::conn_actor::{run_connection, ConnLimits}; +use crate::net::{accept_nonblocking, bind_and_listen, OwnedFd}; +use crate::plug::Pipeline; + +use std::io::{self, ErrorKind}; +use std::net::{SocketAddr, ToSocketAddrs}; +use std::os::fd::RawFd; +use std::time::Duration; + +// --------------------------------------------------------------------------- +// Config +// --------------------------------------------------------------------------- + +#[derive(Clone, Debug)] +pub struct Config { + pub addr: SocketAddr, + pub listener_pool: usize, + pub keep_alive_timeout: Duration, + pub max_header_count: usize, + pub read_buf_size: usize, + pub request_timeout: Duration, + pub max_body_bytes: usize, + /// Number of smarm scheduler OS threads. `None` means smarm's default + /// (one per CPU). Set this to a small fixed number in tests so multiple + /// concurrent test servers don't oversubscribe the host. + pub scheduler_threads: Option, +} + +impl Config { + pub fn new(addr: SocketAddr) -> Self { + let pool = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(2) + .max(2); + Self { + addr, + listener_pool: pool, + keep_alive_timeout: Duration::from_secs(60), + max_header_count: 64, + read_buf_size: 8 * 1024, + request_timeout: Duration::from_secs(30), + max_body_bytes: 16 * 1024 * 1024, + scheduler_threads: None, + } + } + + fn to_conn_limits(&self) -> ConnLimits { + ConnLimits { + max_headers: self.max_header_count, + initial_read_buf: self.read_buf_size, + max_head_bytes: 64 * 1024, + max_body_bytes: self.max_body_bytes, + keep_alive_timeout: self.keep_alive_timeout, + } + } +} + +// --------------------------------------------------------------------------- +// dup helper +// --------------------------------------------------------------------------- + +fn dup_fd(fd: RawFd) -> io::Result { + let new_fd = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) }; + if new_fd < 0 { + return Err(io::Error::last_os_error()); + } + Ok(OwnedFd::from_raw(new_fd)) +} + +// --------------------------------------------------------------------------- +// listener actor body +// --------------------------------------------------------------------------- + +fn listener_loop(listener: OwnedFd, pipeline: Pipeline, limits: ConnLimits) { + let fd = listener.as_raw(); + loop { + match accept_nonblocking(fd) { + Ok(client) => { + // Hand the fd off to a new connection actor. spawn() is + // cheap on smarm — it's a single Vec push under the + // shared lock. + let p = pipeline.clone(); + let l = limits; + smarm::spawn(move || run_connection(client, p, l)); + } + Err(e) if e.kind() == ErrorKind::WouldBlock => { + // No pending connection. Park until the listener is + // readable again, then retry. + if let Err(we) = smarm::wait_readable(fd) { + // epoll registration failed — fatal for this listener. + eprintln!("urus: listener wait_readable failed: {we}"); + return; + } + } + Err(e) if e.kind() == ErrorKind::Interrupted => { + continue; + } + Err(e) => { + // EMFILE / ENFILE / ECONNABORTED etc. Log and continue; + // the system may recover. + eprintln!("urus: accept error: {e}"); + // Small backoff via smarm's sleep to avoid spinning if + // the error is sticky. + smarm::sleep(Duration::from_millis(10)); + } + } + } + // listener OwnedFd drops here, closing the dup'd fd. +} + +// --------------------------------------------------------------------------- +// serve_with — main entry. Boots smarm, spawns listeners, blocks. +// --------------------------------------------------------------------------- +// +// Boots an smarm runtime (one OS thread per CPU by default — see smarm's +// `Config::default()`) and runs until externally killed. We don't wire a +// graceful-shutdown signal in v1; the runtime exits when all actors exit, +// which they don't (listener loops are infinite). Ctrl-C is your friend. + +pub fn serve_with(config: Config, pipeline: Pipeline) -> io::Result<()> { + let listener = bind_and_listen(config.addr)?; + println!("urus: listening on {}", config.addr); + + // We want one connection-actor-spawning loop per listener pool slot. + // Each gets its own dup'd fd so epoll registrations don't collide. + let mut listener_fds = Vec::with_capacity(config.listener_pool); + listener_fds.push(listener); // primary keeps the original + + for _ in 1..config.listener_pool { + let dup = dup_fd(listener_fds[0].as_raw())?; + listener_fds.push(dup); + } + + let limits = config.to_conn_limits(); + + // smarm's runtime API: init(Config) then run(f). The closure is the + // root actor; from there we spawn one listener per fd in the pool. + let smarm_cfg = match config.scheduler_threads { + Some(n) => smarm::Config::exact(n), + None => smarm::Config::default(), + }; + let rt = smarm::init(smarm_cfg); + rt.run(move || { + let n = listener_fds.len(); + let mut handles = Vec::with_capacity(n); + for (i, lfd) in listener_fds.into_iter().enumerate() { + let p = pipeline.clone(); + let h = smarm::spawn(move || { + println!("urus: listener {} starting", i); + listener_loop(lfd, p, limits); + }); + handles.push(h); + } + // Block forever (until ctrl-C) by joining the listeners. They + // never exit on their own in v1. + for h in handles { + let _ = h.join(); + } + }); + + Ok(()) +} + +// --------------------------------------------------------------------------- +// serve — convenience over serve_with. +// --------------------------------------------------------------------------- + +pub fn serve(addr: impl ToSocketAddrs, pipeline: Pipeline) -> io::Result<()> { + let addr = addr + .to_socket_addrs()? + .next() + .ok_or_else(|| io::Error::new(ErrorKind::InvalidInput, "no addresses resolved"))?; + serve_with(Config::new(addr), pipeline) +} diff --git a/tests/integration.rs b/tests/integration.rs new file mode 100644 index 0000000..56e8c51 --- /dev/null +++ b/tests/integration.rs @@ -0,0 +1,227 @@ +//! End-to-end integration tests. +//! +//! Each test spins up the server in a background OS thread on an +//! ephemeral port, opens a regular `std::net::TcpStream` against it, and +//! exercises the wire. Single thread for the smarm runtime per test so +//! teardown happens cleanly when the test thread is dropped (the process +//! exits at the end of the test binary). + +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::time::Duration; + +use urus::{serve_with, Conn, Config, Next, Pipeline, Router}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Reserve a free localhost port by binding briefly and reading back the +/// assigned port number. The port is released the instant we drop the +/// listener; there's a tiny race window before the urus server claims it +/// but it's well below the threshold for flakiness in practice. +fn free_port() -> u16 { + let l = TcpListener::bind("127.0.0.1:0").unwrap(); + l.local_addr().unwrap().port() +} + +fn spawn_server(pipeline: Pipeline) -> 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), + ..Config::new(addr) + }; + serve_with(cfg, pipeline).unwrap(); + }); + // Wait for the server to actually be listening. + 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}"); +} + +fn send_request(port: u16, req: &[u8]) -> Vec { + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + s.write_all(req).unwrap(); + s.shutdown(std::net::Shutdown::Write).ok(); + let mut buf = Vec::new(); + s.read_to_end(&mut buf).unwrap(); + buf +} + +fn http_status(resp: &[u8]) -> u16 { + let s = std::str::from_utf8(&resp[..resp.len().min(64)]).unwrap(); + let parts: Vec<&str> = s.splitn(3, ' ').collect(); + parts[1].parse().unwrap() +} + +fn http_body(resp: &[u8]) -> &[u8] { + // Split on the first \r\n\r\n. + for i in 0..resp.len().saturating_sub(3) { + if &resp[i..i + 4] == b"\r\n\r\n" { + return &resp[i + 4..]; + } + } + &[] +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn hello_world() { + let pipe = Pipeline::new().plug( + Router::new().get("/", |c: Conn, _n: Next| { + c.put_status(200).put_body("hello urus") + }) + ); + let port = spawn_server(pipe); + let resp = send_request(port, b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"hello urus"); +} + +#[test] +fn echo_body() { + let pipe = Pipeline::new().plug( + Router::new().post("/echo", |c: Conn, _n: Next| { + let body = c.body.as_bytes().to_vec(); + c.put_status(200).put_body(body) + }) + ); + let port = spawn_server(pipe); + let resp = send_request( + port, + b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello", + ); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"hello"); +} + +#[test] +fn path_params() { + let pipe = Pipeline::new().plug( + Router::new().get("/users/:id", |c: Conn, _n: Next| { + let id = c.params.get("id").unwrap_or("?").to_string(); + c.put_status(200).put_body(format!("user={id}")) + }) + ); + let port = spawn_server(pipe); + let resp = send_request(port, b"GET /users/42 HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"user=42"); +} + +#[test] +fn method_not_allowed() { + let pipe = Pipeline::new().plug( + Router::new().get("/only-get", |c: Conn, _n: Next| c.put_status(200)) + ); + let port = spawn_server(pipe); + let resp = send_request(port, b"POST /only-get HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 405); +} + +#[test] +fn unknown_route_falls_to_404() { + // Router falls through; connection actor's default emits 404. + let pipe = Pipeline::new().plug( + Router::new().get("/known", |c: Conn, _n: Next| c.put_status(200)) + ); + let port = spawn_server(pipe); + let resp = send_request(port, b"GET /missing HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 404); +} + +#[test] +fn keep_alive_two_requests_on_one_connection() { + let pipe = Pipeline::new().plug( + Router::new().get("/", |c: Conn, _n: Next| c.put_status(200).put_body("ok")) + ); + let port = spawn_server(pipe); + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + + // First request, no Connection header (HTTP/1.1 default = keep-alive). + s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + + // Read the first response (Content-Length: 2 -> body "ok"). + let mut buf = vec![0u8; 1024]; + let mut total = 0; + let mut first_end = None; + while first_end.is_none() { + let n = s.read(&mut buf[total..]).unwrap(); + assert!(n > 0, "server hung up early"); + total += n; + // First response ends at header-terminator + 2 bytes of body. + for i in 0..total.saturating_sub(3) { + if &buf[i..i + 4] == b"\r\n\r\n" { + let body_start = i + 4; + if total >= body_start + 2 { + first_end = Some(body_start + 2); + } + break; + } + } + } + let first = &buf[..first_end.unwrap()]; + assert_eq!(http_status(first), 200); + assert_eq!(http_body(first), b"ok"); + + // Second request on the same socket — proves keep-alive works. + s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap(); + let mut rest = Vec::new(); + s.read_to_end(&mut rest).unwrap(); + assert_eq!(http_status(&rest), 200); + assert_eq!(http_body(&rest), b"ok"); +} + +#[test] +fn panicking_handler_yields_500() { + let pipe = Pipeline::new().plug( + Router::new().get("/boom", |_c: Conn, _n: Next| -> Conn { + panic!("handler explosion") + }) + ); + let port = spawn_server(pipe); + let resp = send_request(port, b"GET /boom HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 500); +} + +#[test] +fn middleware_can_short_circuit() { + // Middleware that requires `X-Auth: secret`; otherwise returns 401 + // without calling `next.run(conn)`. + let auth = |c: Conn, n: Next| { + if c.headers.get("x-auth").map(|s| s == "secret").unwrap_or(false) { + n.run(c) + } else { + c.put_status(401).halt() + } + }; + let pipe = Pipeline::new() + .plug(auth) + .plug(Router::new().get("/secret", |c: Conn, _n: Next| c.put_status(200).put_body("ok"))); + + let port = spawn_server(pipe); + + // Without auth → 401. + let resp = send_request(port, b"GET /secret HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 401); + + // With auth → 200 + body. + let resp = send_request( + port, + b"GET /secret HTTP/1.1\r\nHost: x\r\nX-Auth: secret\r\nConnection: close\r\n\r\n", + ); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"ok"); +}