diff --git a/Cargo.lock b/Cargo.lock index 8c70aa1..9a56072 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,90 +8,12 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" -[[package]] -name = "async-trait" -version = "0.1.89" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "atomic-waker" -version = "1.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" - -[[package]] -name = "axum" -version = "0.7.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" -dependencies = [ - "async-trait", - "axum-core", - "bytes", - "futures-util", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "itoa", - "matchit", - "memchr", - "mime", - "percent-encoding", - "pin-project-lite", - "rustversion", - "serde", - "serde_json", - "serde_path_to_error", - "serde_urlencoded", - "sync_wrapper", - "tokio", - "tower 0.5.3", - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "axum-core" -version = "0.4.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199" -dependencies = [ - "async-trait", - "bytes", - "futures-util", - "http", - "http-body", - "http-body-util", - "mime", - "pin-project-lite", - "rustversion", - "sync_wrapper", - "tower-layer", - "tower-service", - "tracing", -] - [[package]] name = "bitflags" version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" -[[package]] -name = "bytes" -version = "1.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" - [[package]] name = "cfg-if" version = "1.0.4" @@ -126,48 +48,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" -[[package]] -name = "form_urlencoded" -version = "1.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" -dependencies = [ - "percent-encoding", -] - -[[package]] -name = "futures-channel" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" -dependencies = [ - "futures-core", -] - -[[package]] -name = "futures-core" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" - -[[package]] -name = "futures-task" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" - -[[package]] -name = "futures-util" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" -dependencies = [ - "futures-core", - "futures-task", - "pin-project-lite", - "slab", -] - [[package]] name = "getrandom" version = "0.4.2" @@ -202,86 +82,6 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "http" -version = "1.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" -dependencies = [ - "bytes", - "itoa", -] - -[[package]] -name = "http-body" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" -dependencies = [ - "bytes", - "http", -] - -[[package]] -name = "http-body-util" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" -dependencies = [ - "bytes", - "futures-core", - "http", - "http-body", - "pin-project-lite", -] - -[[package]] -name = "httparse" -version = "1.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" - -[[package]] -name = "httpdate" -version = "1.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" - -[[package]] -name = "hyper" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" -dependencies = [ - "atomic-waker", - "bytes", - "futures-channel", - "futures-core", - "http", - "http-body", - "httparse", - "httpdate", - "itoa", - "pin-project-lite", - "smallvec", - "tokio", -] - -[[package]] -name = "hyper-util" -version = "0.1.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" -dependencies = [ - "bytes", - "http", - "http-body", - "hyper", - "pin-project-lite", - "tokio", - "tower-service", -] - [[package]] name = "id-arena" version = "2.3.0" @@ -330,53 +130,18 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" -[[package]] -name = "matchit" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" - [[package]] name = "memchr" version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" -[[package]] -name = "mime" -version = "0.3.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" - -[[package]] -name = "mio" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" -dependencies = [ - "libc", - "wasi", - "windows-sys", -] - [[package]] name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" -[[package]] -name = "percent-encoding" -version = "2.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" - -[[package]] -name = "pin-project-lite" -version = "0.2.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" - [[package]] name = "prettyplease" version = "0.2.37" @@ -416,13 +181,10 @@ name = "rt" version = "0.1.0" dependencies = [ "anyhow", - "axum", "libc", "serde", "serde_json", "tempfile", - "tokio", - "tower 0.4.13", ] [[package]] @@ -438,18 +200,6 @@ dependencies = [ "windows-sys", ] -[[package]] -name = "rustversion" -version = "1.0.22" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" - -[[package]] -name = "ryu" -version = "1.0.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" - [[package]] name = "semver" version = "1.0.27" @@ -499,61 +249,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "serde_path_to_error" -version = "0.1.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" -dependencies = [ - "itoa", - "serde", - "serde_core", -] - -[[package]] -name = "serde_urlencoded" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" -dependencies = [ - "form_urlencoded", - "itoa", - "ryu", - "serde", -] - -[[package]] -name = "signal-hook-registry" -version = "1.4.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" -dependencies = [ - "errno", - "libc", -] - -[[package]] -name = "slab" -version = "0.4.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" - -[[package]] -name = "smallvec" -version = "1.15.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" - -[[package]] -name = "socket2" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" -dependencies = [ - "libc", - "windows-sys", -] - [[package]] name = "syn" version = "2.0.117" @@ -565,12 +260,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "sync_wrapper" -version = "1.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" - [[package]] name = "tempfile" version = "3.27.0" @@ -584,91 +273,6 @@ dependencies = [ "windows-sys", ] -[[package]] -name = "tokio" -version = "1.52.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" -dependencies = [ - "libc", - "mio", - "pin-project-lite", - "signal-hook-registry", - "socket2", - "tokio-macros", - "windows-sys", -] - -[[package]] -name = "tokio-macros" -version = "2.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "tower" -version = "0.4.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" -dependencies = [ - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "tower" -version = "0.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" -dependencies = [ - "futures-core", - "futures-util", - "pin-project-lite", - "sync_wrapper", - "tokio", - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "tower-layer" -version = "0.3.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" - -[[package]] -name = "tower-service" -version = "0.3.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" - -[[package]] -name = "tracing" -version = "0.1.44" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" -dependencies = [ - "log", - "pin-project-lite", - "tracing-core", -] - -[[package]] -name = "tracing-core" -version = "0.1.36" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" -dependencies = [ - "once_cell", -] - [[package]] name = "unicode-ident" version = "1.0.24" @@ -681,12 +285,6 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" -[[package]] -name = "wasi" -version = "0.11.1+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" - [[package]] name = "wasip2" version = "1.0.2+wasi-0.2.9" diff --git a/crates/rt/Cargo.toml b/crates/rt/Cargo.toml index 5508e1a..5b5694f 100644 --- a/crates/rt/Cargo.toml +++ b/crates/rt/Cargo.toml @@ -16,9 +16,6 @@ path = "src/bin/wo.rs" anyhow = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" -tokio = { version = "1", features = ["rt", "macros", "net", "signal", "sync", "time"] } -axum = "0.7" -tower = "0.4" libc = "0.2" # phase 02 — direct epoll/eventfd/timerfd/signalfd syscalls [dev-dependencies] diff --git a/crates/rt/src/bin/wo.rs b/crates/rt/src/bin/wo.rs index 739e99b..4c62830 100644 --- a/crates/rt/src/bin/wo.rs +++ b/crates/rt/src/bin/wo.rs @@ -1,15 +1,23 @@ //! `wo` — the writeonce toolchain binary. //! //! Stage 2 scope: -//! wo run — discover `.wo` files under , parse the type DSL, -//! compile a catalog, and serve REST CRUD on :8080. -//! wo --help — print usage. +//! wo run discover `.wo` files under , parse the type DSL, +//! compile a catalog, and serve REST CRUD on :8080. +//! wo --help print usage. +//! +//! After the phase-04 cutover this binary owns one event loop on one +//! thread. No tokio. The same pattern Redis and TigerBeetle use — see +//! `docs/runtime/database/02-wo-language.md § Concurrency Model`. +use std::collections::HashMap; +use std::os::unix::io::{AsRawFd, RawFd}; use std::path::PathBuf; use std::process::ExitCode; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; +use std::time::Duration; -use tokio::sync::Mutex; +use rt::http::{Connection, Listener, Router}; +use rt::runtime::{EventLoop, Interest, SignalFd, Token}; fn usage() { eprintln!( @@ -30,38 +38,22 @@ fn main() -> ExitCode { let args: Vec = std::env::args().skip(1).collect(); let slice: Vec<&str> = args.iter().map(|s| s.as_str()).collect(); match slice.as_slice() { - [] | ["--help"] | ["-h"] => { - usage(); - ExitCode::from(0) - } + [] | ["--help"] | ["-h"] => { usage(); ExitCode::from(0) } ["run"] => { eprintln!("wo run: directory argument required"); usage(); ExitCode::from(2) } - ["run", dir] => run(PathBuf::from(dir)), - _ => { - usage(); - ExitCode::from(2) - } + ["run", dir] => match run(PathBuf::from(dir)) { + Ok(c) => c, + Err(e) => { eprintln!("error: {e}"); ExitCode::from(1) } + }, + _ => { usage(); ExitCode::from(2) } } } -fn run(dir: PathBuf) -> ExitCode { - // Tokio current-thread runtime — matches the single-threaded engine design. - let rt = match tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - { - Ok(r) => r, - Err(e) => { eprintln!("error: build runtime: {e}"); return ExitCode::from(1); } - }; - rt.block_on(async move { serve(dir).await }) - .unwrap_or_else(|e| { eprintln!("error: {e}"); ExitCode::from(1) }) -} - -async fn serve(dir: PathBuf) -> anyhow::Result { - // 1. Discover +fn run(dir: PathBuf) -> anyhow::Result { + // 1. Discover .wo files. let files = rt::discover(&dir)?; if files.is_empty() { anyhow::bail!("no .wo files found under {}", dir.display()); @@ -72,11 +64,11 @@ async fn serve(dir: PathBuf) -> anyhow::Result { dir.display(), ); - // 2. Parse each file into a Schema + // 2. Parse each into a Schema. let mut schemas = Vec::new(); for f in &files { match rt::parser::parse(&f.src) { - Ok(s) => { + Ok(s) => { let n = s.types.len(); println!(" parsed {} — {} type{}", f.rel.display(), n, if n == 1 { "" } else { "s" }); schemas.push(s); @@ -88,39 +80,95 @@ async fn serve(dir: PathBuf) -> anyhow::Result { } } - // 3. Compile a catalog + // 3. Compile the catalog. let catalog = rt::compile::Catalog::from_schemas(schemas)?; println!("[wo] compiled catalog — {} type{}", catalog.order.len(), if catalog.order.len() == 1 { "" } else { "s" }); - // 4. Boot the engine + // 4. Boot the engine + the router. let engine = Arc::new(Mutex::new(rt::engine::Engine::new(catalog.clone()))); + println!(); + println!("[wo] routes:"); { - let e = engine.lock().await; - println!(); - println!("[wo] routes:"); + let e = engine.lock().unwrap(); print!("{}", rt::server::describe_routes(&e)); } + let router = rt::server::router(engine.clone(), &catalog); - // 5. Bind and serve + // 5. Bind and serve. let addr = std::env::var("WO_LISTEN").unwrap_or_else(|_| "127.0.0.1:8080".to_string()); - let listener = match tokio::net::TcpListener::bind(&addr).await { - Ok(l) => l, - Err(e) => anyhow::bail!("bind {addr}: {e}"), - }; + let listener = Listener::bind(&addr) + .map_err(|e| anyhow::anyhow!("bind {addr}: {e}"))?; println!(); - println!("[wo] listening on http://{addr}"); + println!("[wo] listening on http://{}", listener.local_addr()); println!("[wo] ctrl-C to stop"); - let app = rt::server::router(engine.clone(), &catalog); - - // Graceful shutdown on ctrl-C - axum::serve(listener, app) - .with_graceful_shutdown(async { - let _ = tokio::signal::ctrl_c().await; - println!("\n[wo] shutting down"); - }) - .await?; - + serve_loop(listener, router)?; Ok(ExitCode::from(0)) } + +fn serve_loop(listener: Listener, router: Router) -> anyhow::Result<()> { + let mut eloop = EventLoop::new()?; + let signals = SignalFd::new()?; + let listen_fd = listener.as_raw_fd(); + let signal_fd = signals.as_raw_fd(); + + // Tokens: connection fds carry their own raw fd as the token; the + // listener and signalfd use their fds too — they're disjoint by + // construction (different fds). + eloop.register(listen_fd, Interest::READABLE, Token(listen_fd as u64))?; + eloop.register(signal_fd, Interest::READABLE, Token(signal_fd as u64))?; + + let mut conns: HashMap = HashMap::new(); + + 'outer: loop { + let events = match eloop.wait_once(Some(Duration::from_secs(60))) { + Ok(evs) => evs, + Err(e) => { + eprintln!("[wo] event loop error: {e}"); + continue; + } + }; + + for ev in events { + let fd = ev.token().0 as RawFd; + + if fd == listen_fd { + // Drain accept queue (edge-triggered). + while let Some(cfd) = listener.accept()? { + eloop.register(cfd, Interest::READABLE, Token(cfd as u64))?; + conns.insert(cfd, Connection::new(cfd)); + } + continue; + } + + if fd == signal_fd { + let sig = signals.read().unwrap_or(0); + println!(); + println!("[wo] received signal {sig} — shutting down"); + break 'outer; + } + + // Connection event. + let Some(conn) = conns.get_mut(&fd) else { continue }; + let want_writable = match conn.drive(ev.readable, ev.writable, ev.hangup, ev.error, &router) { + Ok(w) => w, + Err(_) => { conns.remove(&fd); continue; } + }; + + if conn.is_done() { + eloop.deregister(fd).ok(); + conns.remove(&fd); // Drop closes the fd. + } else if want_writable { + let _ = eloop.modify(fd, Interest::READ_WRITE, Token(fd as u64)); + } + } + } + + // Tear down outstanding connections cleanly. Dropping Connection closes + // each fd; deregistering from the loop is optional (close auto-removes). + for (fd, _) in conns.drain() { + let _ = eloop.deregister(fd); + } + Ok(()) +} diff --git a/crates/rt/src/http/connection.rs b/crates/rt/src/http/connection.rs new file mode 100644 index 0000000..5abdbf7 --- /dev/null +++ b/crates/rt/src/http/connection.rs @@ -0,0 +1,247 @@ +//! Per-connection state machine driven by the phase-02 [`EventLoop`]. +//! +//! Lifecycle (close-after-response, no keep-alive yet): +//! Reading → drain `read(2)` to `EAGAIN`, parse, dispatch through the +//! `Router`, queue the response. +//! Writing → drain `write(2)` to `EAGAIN`. If a write was partial, the +//! loop re-arms the fd as `WRITABLE` and we continue on the next event. +//! Done → loop closes the fd. +//! +//! Adapted from `reference/crates/wo-http/src/connection.rs`. The owning +//! [`EventLoop`] supplies `read`/`write` readiness via edge-triggered +//! `epoll`; this struct is the per-fd part of the state. +//! +//! [`EventLoop`]: crate::runtime::EventLoop + +use std::io; +use std::os::unix::io::{AsRawFd, RawFd}; + +use super::request::{self, ParseResult}; +use super::response::Response; +use super::route::Router; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ConnState { + Reading, + Writing, + Done, +} + +pub struct Connection { + fd: RawFd, + state: ConnState, + read_buf: Vec, + write_buf: Vec, + write_offset: usize, +} + +impl Connection { + pub fn new(fd: RawFd) -> Self { + Self { + fd, + state: ConnState::Reading, + read_buf: Vec::with_capacity(4096), + write_buf: Vec::new(), + write_offset: 0, + } + } + + pub fn state(&self) -> ConnState { self.state } + pub fn is_done(&self) -> bool { self.state == ConnState::Done } + + /// Drain the socket into `read_buf` until `EAGAIN` or EOF. + /// Returns `false` when the peer closed (connection should be torn down). + fn drain_read(&mut self) -> io::Result { + let mut tmp = [0u8; 4096]; + loop { + let n = unsafe { + libc::read(self.fd, tmp.as_mut_ptr() as *mut libc::c_void, tmp.len()) + }; + if n < 0 { + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::EAGAIN) { + return Ok(true); + } + return Err(err); + } + if n == 0 { + return Ok(false); + } + self.read_buf.extend_from_slice(&tmp[..n as usize]); + // Cap at MAX_BODY_BYTES + headers — refuse pathological requests. + if self.read_buf.len() > 32 * 1024 * 1024 { + return Err(io::Error::new(io::ErrorKind::InvalidData, "request too large")); + } + } + } + + /// Drain `write_buf[write_offset..]` to the socket until `EAGAIN`. + /// Returns `true` once everything has been flushed. + fn drain_write(&mut self) -> io::Result { + loop { + let remaining = &self.write_buf[self.write_offset..]; + if remaining.is_empty() { return Ok(true); } + let n = unsafe { + libc::write( + self.fd, + remaining.as_ptr() as *const libc::c_void, + remaining.len(), + ) + }; + if n < 0 { + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::EAGAIN) { + return Ok(false); + } + return Err(err); + } + if n == 0 { return Ok(false); } + self.write_offset += n as usize; + } + } + + fn try_parse(&self) -> ParseResult { + request::parse(&self.read_buf) + } + + fn queue_response(&mut self, response: &Response) { + self.write_buf = response.to_bytes(); + self.write_offset = 0; + self.state = ConnState::Writing; + } + + /// One step of the state machine, given a readiness event from the + /// loop. Returns `true` if the connection now wants `WRITABLE` (the + /// caller should switch interest from `READABLE`); `false` otherwise. + pub fn drive( + &mut self, + readable: bool, + writable: bool, + hangup: bool, + error: bool, + router: &Router, + ) -> io::Result { + if error { + self.state = ConnState::Done; + return Ok(false); + } + + if readable && self.state == ConnState::Reading { + let still_open = self.drain_read()?; + match self.try_parse() { + ParseResult::Complete(req) => { + let resp = router.dispatch(&req); + self.queue_response(&resp); + } + ParseResult::Incomplete => { + if !still_open { + self.state = ConnState::Done; + return Ok(false); + } + } + ParseResult::Error(msg) => { + let resp = Response::status(super::Status::BAD_REQUEST).text(msg); + self.queue_response(&resp); + } + } + } + + if self.state == ConnState::Writing { + let flushed = self.drain_write()?; + if flushed { + self.state = ConnState::Done; + return Ok(false); + } else if !writable { + // We tried, EAGAIN'd; tell the caller to wait for WRITABLE. + return Ok(true); + } + } + + if hangup && self.state != ConnState::Writing { + self.state = ConnState::Done; + } + + Ok(false) + } +} + +impl AsRawFd for Connection { + fn as_raw_fd(&self) -> RawFd { self.fd } +} + +impl Drop for Connection { + fn drop(&mut self) { + unsafe { libc::close(self.fd); } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::http::{Method, Response, Router}; + + fn socketpair_nonblock() -> (RawFd, RawFd) { + let mut fds = [0i32; 2]; + let r = unsafe { + libc::socketpair( + libc::AF_UNIX, + libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + fds.as_mut_ptr(), + ) + }; + assert_eq!(r, 0); + (fds[0], fds[1]) + } + + #[test] + fn connection_handles_a_request() { + let (server_fd, client_fd) = socketpair_nonblock(); + + let req = b"GET /healthz HTTP/1.1\r\nHost: localhost\r\n\r\n"; + let n = unsafe { + libc::write(client_fd, req.as_ptr() as *const _, req.len()) + }; + assert_eq!(n, req.len() as isize); + + let router = Router::new() + .route(Method::Get, "/healthz", |_, _| Response::ok().text("ok")); + + let mut conn = Connection::new(server_fd); + let want_writable = conn.drive(true, false, false, false, &router).unwrap(); + assert!(!want_writable, "small response fits in one write"); + assert!(conn.is_done()); + + let mut buf = [0u8; 4096]; + let n = unsafe { + libc::read(client_fd, buf.as_mut_ptr() as *mut _, buf.len()) + }; + assert!(n > 0); + let s = std::str::from_utf8(&buf[..n as usize]).unwrap(); + assert!(s.starts_with("HTTP/1.1 200 OK\r\n"), "got: {s}"); + assert!(s.ends_with("\r\n\r\nok")); + + unsafe { libc::close(client_fd); } + // server_fd is closed by Connection::drop. + } + + #[test] + fn returns_404_for_unknown_path() { + let (server_fd, client_fd) = socketpair_nonblock(); + + let req = b"GET /missing HTTP/1.1\r\n\r\n"; + unsafe { libc::write(client_fd, req.as_ptr() as *const _, req.len()); } + + let router = Router::new() + .route(Method::Get, "/healthz", |_, _| Response::ok().text("ok")); + let mut conn = Connection::new(server_fd); + conn.drive(true, false, false, false, &router).unwrap(); + + let mut buf = [0u8; 4096]; + let n = unsafe { libc::read(client_fd, buf.as_mut_ptr() as *mut _, buf.len()) }; + let s = std::str::from_utf8(&buf[..n as usize]).unwrap(); + assert!(s.starts_with("HTTP/1.1 404 Not Found\r\n"), "got: {s}"); + + unsafe { libc::close(client_fd); } + } +} diff --git a/crates/rt/src/http/listener.rs b/crates/rt/src/http/listener.rs new file mode 100644 index 0000000..7fd4fc8 --- /dev/null +++ b/crates/rt/src/http/listener.rs @@ -0,0 +1,150 @@ +//! Non-blocking TCP listener — `socket(2)` + `bind(2)` + `listen(2)` + `accept4(2)`. +//! +//! Adapted from `reference/crates/wo-http/src/listener.rs`. The v1 hand-rolled +//! IPv4 parser had a byte-order bug for non-localhost addresses; here we +//! defer to `std::net::SocketAddr` (stdlib, no extra crate) and convert the +//! resulting octets to a `sockaddr_in` correctly. + +use std::io; +use std::net::SocketAddr; +use std::os::unix::io::{AsRawFd, RawFd}; + +pub struct Listener { + fd: RawFd, + addr: SocketAddr, +} + +impl Listener { + /// Bind to `addr` (IPv4 only for now) and start listening with backlog 128. + /// Socket is created `SOCK_NONBLOCK | SOCK_CLOEXEC`. + pub fn bind(addr: &str) -> io::Result { + let parsed: SocketAddr = addr.parse().map_err(|e| { + io::Error::new(io::ErrorKind::InvalidInput, format!("bad addr {addr:?}: {e}")) + })?; + let SocketAddr::V4(v4) = parsed else { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "IPv4 only for now")); + }; + + let fd = unsafe { + libc::socket( + libc::AF_INET, + libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + ) + }; + if fd < 0 { return Err(io::Error::last_os_error()); } + + // SO_REUSEADDR so the same port restarts cleanly between `wo run`s. + let one: libc::c_int = 1; + let ret = unsafe { + libc::setsockopt( + fd, libc::SOL_SOCKET, libc::SO_REUSEADDR, + &one as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + let err = io::Error::last_os_error(); + unsafe { libc::close(fd); } + return Err(err); + } + + let s_addr = u32::from_be_bytes(v4.ip().octets()).to_be(); + let sock = libc::sockaddr_in { + sin_family: libc::AF_INET as libc::sa_family_t, + sin_port: v4.port().to_be(), + sin_addr: libc::in_addr { s_addr }, + sin_zero: [0; 8], + }; + let ret = unsafe { + libc::bind( + fd, + &sock as *const _ as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + let err = io::Error::last_os_error(); + unsafe { libc::close(fd); } + return Err(err); + } + + if unsafe { libc::listen(fd, 128) } < 0 { + let err = io::Error::last_os_error(); + unsafe { libc::close(fd); } + return Err(err); + } + + // Resolve the actual bound address — caller may have asked for port 0. + let local = read_local_addr(fd)?; + Ok(Self { fd, addr: local }) + } + + /// Accept the next pending connection. Returns `None` on `EAGAIN`. + pub fn accept(&self) -> io::Result> { + let cfd = unsafe { + libc::accept4( + self.fd, + std::ptr::null_mut(), std::ptr::null_mut(), + libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + ) + }; + if cfd < 0 { + let err = io::Error::last_os_error(); + // On Linux EAGAIN == EWOULDBLOCK; one branch is enough. + return match err.raw_os_error() { + Some(libc::EAGAIN) => Ok(None), + _ => Err(err), + }; + } + Ok(Some(cfd)) + } + + pub fn local_addr(&self) -> SocketAddr { self.addr } +} + +impl AsRawFd for Listener { + fn as_raw_fd(&self) -> RawFd { self.fd } +} + +impl Drop for Listener { + fn drop(&mut self) { + unsafe { libc::close(self.fd); } + } +} + +fn read_local_addr(fd: RawFd) -> io::Result { + let mut sock: libc::sockaddr_in = unsafe { std::mem::zeroed() }; + let mut len = std::mem::size_of::() as libc::socklen_t; + let ret = unsafe { + libc::getsockname(fd, &mut sock as *mut _ as *mut libc::sockaddr, &mut len) + }; + if ret < 0 { return Err(io::Error::last_os_error()); } + let ip = u32::from_be(sock.sin_addr.s_addr).to_be_bytes(); + let port = u16::from_be(sock.sin_port); + Ok(SocketAddr::from(([ip[0], ip[1], ip[2], ip[3]], port))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn binds_and_accepts_a_client() { + let listener = Listener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().port(); + assert!(port > 0); + assert!(listener.accept().unwrap().is_none(), "no clients pending yet"); + + let stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + // Block briefly until the server-side accept sees it. + let mut accepted = None; + for _ in 0..50 { + if let Some(fd) = listener.accept().unwrap() { accepted = Some(fd); break; } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + let cfd = accepted.expect("accept produced a client fd"); + unsafe { libc::close(cfd); } + drop(stream); + } +} diff --git a/crates/rt/src/http/mod.rs b/crates/rt/src/http/mod.rs new file mode 100644 index 0000000..5a412c3 --- /dev/null +++ b/crates/rt/src/http/mod.rs @@ -0,0 +1,21 @@ +//! Hand-rolled HTTP/1.1 — phase 03 of the runtime plan. +//! +//! The transport layer for the `wo` binary after the phase-04 cutover. +//! Drives non-blocking accept + per-connection state machines off the +//! phase-02 [`EventLoop`](super::runtime::EventLoop). No `tokio`, no +//! `axum`, no `hyper`. Synchronous handlers; close-after-response (the +//! v1 model — keep-alive lands when a sample needs it). +//! +//! See `docs/plan/03-hand-rolled-http.md` and `docs/plan/04-cutover-remove-tokio-axum.md`. + +mod connection; +mod listener; +mod request; +mod response; +mod route; + +pub use connection::{Connection, ConnState}; +pub use listener::Listener; +pub use request::{Method, Request}; +pub use response::{Response, Status}; +pub use route::{RouteParams, Router}; diff --git a/crates/rt/src/http/request.rs b/crates/rt/src/http/request.rs new file mode 100644 index 0000000..2d0d17a --- /dev/null +++ b/crates/rt/src/http/request.rs @@ -0,0 +1,179 @@ +//! Incremental HTTP/1.1 request parser. +//! +//! Adapted from `reference/crates/wo-http/src/request.rs`. v1 only parsed +//! request headers (the v1 blog is read-only HTML). The phase-04 cutover +//! needs JSON request bodies, so this parser also drains a +//! `Content-Length`-delimited body. Chunked transfer encoding is not +//! supported (no sample sends one — see `docs/plan/03-hand-rolled-http.md`). + +use std::collections::HashMap; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum Method { + Get, + Post, + Patch, + Put, + Delete, + Head, + Options, + Other, +} + +impl Method { + fn parse(token: &str) -> Method { + match token { + "GET" => Method::Get, + "POST" => Method::Post, + "PATCH" => Method::Patch, + "PUT" => Method::Put, + "DELETE" => Method::Delete, + "HEAD" => Method::Head, + "OPTIONS" => Method::Options, + _ => Method::Other, + } + } +} + +#[derive(Debug, Clone)] +pub struct Request { + pub method: Method, + pub path: String, + pub query: Option, + pub headers: HashMap, + pub body: Vec, +} + +pub enum ParseResult { + Complete(Request), + Incomplete, + Error(String), +} + +const MAX_HEADER_BYTES: usize = 8 * 1024; +const MAX_BODY_BYTES: usize = 16 * 1024 * 1024; + +/// Try to parse a complete HTTP/1.1 request out of `buf`. +/// Returns `Complete(req, bytes_consumed)` once headers + body are present. +pub fn parse(buf: &[u8]) -> ParseResult { + let header_end = match find_header_end(buf) { + Some(p) => p, + None => { + if buf.len() > MAX_HEADER_BYTES { + return ParseResult::Error("request headers too large".into()); + } + return ParseResult::Incomplete; + } + }; + + let header_str = match std::str::from_utf8(&buf[..header_end]) { + Ok(s) => s, + Err(_) => return ParseResult::Error("non-UTF-8 in headers".into()), + }; + + let mut lines = header_str.lines(); + + let request_line = match lines.next() { + Some(l) => l, + None => return ParseResult::Error("empty request".into()), + }; + let mut parts = request_line.split_whitespace(); + let method = match parts.next() { + Some(t) => Method::parse(t), + None => return ParseResult::Error("missing method".into()), + }; + let raw_path = match parts.next() { + Some(p) => p, + None => return ParseResult::Error("missing path".into()), + }; + let (path, query) = match raw_path.split_once('?') { + Some((p, q)) => (p.to_string(), Some(q.to_string())), + None => (raw_path.to_string(), None), + }; + + let mut headers = HashMap::new(); + for line in lines { + if line.is_empty() { break; } + if let Some((k, v)) = line.split_once(':') { + headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string()); + } + } + + let body_len: usize = headers + .get("content-length") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + if body_len > MAX_BODY_BYTES { + return ParseResult::Error("Content-Length exceeds limit".into()); + } + + let header_bytes = header_end + 4; // include the trailing \r\n\r\n + let total = header_bytes + body_len; + if buf.len() < total { + return ParseResult::Incomplete; + } + + let body = buf[header_bytes..total].to_vec(); + ParseResult::Complete(Request { method, path, query, headers, body }) +} + +fn find_header_end(buf: &[u8]) -> Option { + buf.windows(4).position(|w| w == b"\r\n\r\n") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_simple_get() { + let raw = b"GET /api/articles HTTP/1.1\r\nHost: localhost\r\n\r\n"; + let ParseResult::Complete(req) = parse(raw) else { panic!("expected Complete") }; + assert_eq!(req.method, Method::Get); + assert_eq!(req.path, "/api/articles"); + assert!(req.query.is_none()); + assert!(req.body.is_empty()); + } + + #[test] + fn parses_query_string() { + let raw = b"GET /tag/rust?page=2 HTTP/1.1\r\n\r\n"; + let ParseResult::Complete(req) = parse(raw) else { panic!() }; + assert_eq!(req.path, "/tag/rust"); + assert_eq!(req.query.as_deref(), Some("page=2")); + } + + #[test] + fn parses_post_with_body() { + let body = b"{\"title\":\"hi\"}"; + let mut raw = Vec::new(); + raw.extend_from_slice(b"POST /api/articles HTTP/1.1\r\n"); + raw.extend_from_slice(b"Host: localhost\r\n"); + raw.extend_from_slice(b"Content-Type: application/json\r\n"); + raw.extend_from_slice(format!("Content-Length: {}\r\n", body.len()).as_bytes()); + raw.extend_from_slice(b"\r\n"); + raw.extend_from_slice(body); + let ParseResult::Complete(req) = parse(&raw) else { panic!() }; + assert_eq!(req.method, Method::Post); + assert_eq!(req.body, body); + } + + #[test] + fn incomplete_when_body_truncated() { + let raw = b"POST / HTTP/1.1\r\nContent-Length: 10\r\n\r\nshort"; + assert!(matches!(parse(raw), ParseResult::Incomplete)); + } + + #[test] + fn incomplete_when_headers_truncated() { + let raw = b"GET / HTTP/1.1\r\nHost: local"; + assert!(matches!(parse(raw), ParseResult::Incomplete)); + } + + #[test] + fn parses_patch_method() { + let raw = b"PATCH /api/articles/1 HTTP/1.1\r\nContent-Length: 0\r\n\r\n"; + let ParseResult::Complete(req) = parse(raw) else { panic!() }; + assert_eq!(req.method, Method::Patch); + } +} diff --git a/crates/rt/src/http/response.rs b/crates/rt/src/http/response.rs new file mode 100644 index 0000000..707cfd3 --- /dev/null +++ b/crates/rt/src/http/response.rs @@ -0,0 +1,113 @@ +//! HTTP/1.1 response builder + serializer. +//! +//! Adapted from `reference/crates/wo-http/src/response.rs`. Adds: +//! * `Status` constants for the codes the REST samples assert on +//! (200/201/204/400/404/405/500/501). +//! * `Response::json(&serde_json::Value)` matching the cutover-handler +//! shape in `docs/plan/04-cutover-remove-tokio-axum.md`. + +use serde_json::Value; + +#[derive(Debug, Clone, Copy)] +pub struct Status(pub u16, pub &'static str); + +impl Status { + pub const OK: Status = Status(200, "OK"); + pub const CREATED: Status = Status(201, "Created"); + pub const NO_CONTENT: Status = Status(204, "No Content"); + pub const BAD_REQUEST: Status = Status(400, "Bad Request"); + pub const NOT_FOUND: Status = Status(404, "Not Found"); + pub const METHOD_NOT_ALLOWED: Status = Status(405, "Method Not Allowed"); + pub const INTERNAL_SERVER_ERROR: Status = Status(500, "Internal Server Error"); + pub const NOT_IMPLEMENTED: Status = Status(501, "Not Implemented"); +} + +#[derive(Debug, Clone)] +pub struct Response { + pub status: Status, + pub headers: Vec<(String, String)>, + pub body: Vec, +} + +impl Response { + pub fn status(s: Status) -> Self { + Self { status: s, headers: Vec::new(), body: Vec::new() } + } + + pub fn ok() -> Self { Self::status(Status::OK) } + pub fn created() -> Self { Self::status(Status::CREATED) } + pub fn no_content() -> Self { Self::status(Status::NO_CONTENT) } + + pub fn header(mut self, k: &str, v: &str) -> Self { + self.headers.push((k.to_string(), v.to_string())); + self + } + + pub fn body(mut self, body: impl Into>) -> Self { + self.body = body.into(); + self + } + + /// Plain-text body with `Content-Type: text/plain; charset=utf-8`. + pub fn text(self, body: impl Into) -> Self { + let body: String = body.into(); + self.header("Content-Type", "text/plain; charset=utf-8") + .body(body.into_bytes()) + } + + /// JSON body with `Content-Type: application/json`. + pub fn json(self, value: &Value) -> Self { + let buf = serde_json::to_vec(value).unwrap_or_else(|_| b"null".to_vec()); + self.header("Content-Type", "application/json").body(buf) + } + + /// Serialize to the wire format. Auto-injects `Content-Length` and + /// `Connection: close` (no keep-alive in Stage 2). + pub fn to_bytes(&self) -> Vec { + let mut buf = Vec::with_capacity(256 + self.body.len()); + buf.extend_from_slice( + format!("HTTP/1.1 {} {}\r\n", self.status.0, self.status.1).as_bytes(), + ); + for (k, v) in &self.headers { + buf.extend_from_slice(format!("{k}: {v}\r\n").as_bytes()); + } + buf.extend_from_slice(format!("Content-Length: {}\r\n", self.body.len()).as_bytes()); + buf.extend_from_slice(b"Connection: close\r\n"); + buf.extend_from_slice(b"\r\n"); + buf.extend_from_slice(&self.body); + buf + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn ok_text_body() { + let r = Response::ok().text("ok"); + let s = String::from_utf8(r.to_bytes()).unwrap(); + assert!(s.starts_with("HTTP/1.1 200 OK\r\n")); + assert!(s.contains("Content-Type: text/plain")); + assert!(s.contains("Content-Length: 2\r\n")); + assert!(s.ends_with("\r\n\r\nok")); + } + + #[test] + fn json_body() { + let r = Response::created().json(&json!({"id": 1, "title": "Hi"})); + let s = String::from_utf8(r.to_bytes()).unwrap(); + assert!(s.starts_with("HTTP/1.1 201 Created\r\n")); + assert!(s.contains("Content-Type: application/json")); + assert!(s.contains(r#"{"id":1,"title":"Hi"}"#)); + } + + #[test] + fn no_content_status() { + let r = Response::no_content(); + let s = String::from_utf8(r.to_bytes()).unwrap(); + assert!(s.starts_with("HTTP/1.1 204 No Content\r\n")); + assert!(s.ends_with("\r\n\r\n")); + } +} diff --git a/crates/rt/src/http/route.rs b/crates/rt/src/http/route.rs new file mode 100644 index 0000000..c9a4f0e --- /dev/null +++ b/crates/rt/src/http/route.rs @@ -0,0 +1,182 @@ +//! Method + URL pattern → handler dispatch. +//! +//! Combined adaptation of `reference/crates/wo-route/src/{router,pattern}.rs`. +//! Handler shape is `Fn(&Request, &RouteParams) -> Response`, captured as a +//! boxed closure so each route closes over its own state (typically an +//! `Arc>` — see `crates/rt/src/server.rs`). +//! +//! Dispatch distinguishes 404 (no path matches any registered route) from +//! 405 (path matches at least one route but not for the request's method), +//! which the REST sample asserts (`expose list, get, ...` → POST → 405). + +use std::collections::HashMap; + +use super::{Method, Request, Response, Status}; + +pub type HandlerFn = dyn Fn(&Request, &RouteParams) -> Response + Send + Sync + 'static; + +#[derive(Debug, Clone, PartialEq)] +enum Segment { + Literal(String), + Param(String), + Wildcard(String), +} + +#[derive(Debug, Clone)] +struct Pattern { + segments: Vec, +} + +impl Pattern { + fn compile(s: &str) -> Self { + let segments = s.trim_start_matches('/') + .split('/') + .filter(|s| !s.is_empty()) + .map(|seg| { + if let Some(name) = seg.strip_prefix(':') { + Segment::Param(name.to_string()) + } else if let Some(name) = seg.strip_prefix('*') { + Segment::Wildcard(name.to_string()) + } else { + Segment::Literal(seg.to_string()) + } + }) + .collect(); + Self { segments } + } + + fn matches(&self, path: &str) -> Option> { + let parts: Vec<&str> = path.trim_start_matches('/') + .split('/') + .filter(|s| !s.is_empty()) + .collect(); + let mut params = Vec::new(); + let mut pi = 0usize; + for seg in &self.segments { + match seg { + Segment::Literal(lit) => { + if pi >= parts.len() || parts[pi] != lit { return None; } + pi += 1; + } + Segment::Param(name) => { + if pi >= parts.len() { return None; } + params.push((name.clone(), parts[pi].to_string())); + pi += 1; + } + Segment::Wildcard(name) => { + if pi >= parts.len() { return None; } + params.push((name.clone(), parts[pi..].join("/"))); + return Some(params); + } + } + } + if pi == parts.len() { Some(params) } else { None } + } +} + +#[derive(Debug, Clone, Default)] +pub struct RouteParams { + params: HashMap, +} + +impl RouteParams { + pub fn get(&self, k: &str) -> Option<&str> { + self.params.get(k).map(String::as_str) + } + + fn from_pairs(pairs: Vec<(String, String)>) -> Self { + Self { params: pairs.into_iter().collect() } + } +} + +struct Route { + method: Method, + pattern: Pattern, + handler: Box, +} + +#[derive(Default)] +pub struct Router { + routes: Vec, +} + +impl Router { + pub fn new() -> Self { Self::default() } + + pub fn route(mut self, method: Method, pattern: &str, handler: F) -> Self + where + F: Fn(&Request, &RouteParams) -> Response + Send + Sync + 'static, + { + self.routes.push(Route { + method, + pattern: Pattern::compile(pattern), + handler: Box::new(handler), + }); + self + } + + /// Resolve a request to a response. 404 if no path matches; 405 if the + /// path matches a registered route under a different method. + pub fn dispatch(&self, req: &Request) -> Response { + let mut path_matched_any = false; + for r in &self.routes { + if let Some(pairs) = r.pattern.matches(&req.path) { + if r.method == req.method { + let params = RouteParams::from_pairs(pairs); + return (r.handler)(req, ¶ms); + } + path_matched_any = true; + } + } + if path_matched_any { + Response::status(Status::METHOD_NOT_ALLOWED).text("method not allowed") + } else { + Response::status(Status::NOT_FOUND).text("no route") + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + fn req(method: Method, path: &str) -> Request { + Request { method, path: path.into(), query: None, headers: HashMap::new(), body: vec![] } + } + + #[test] + fn dispatches_static_and_param_routes() { + let r = Router::new() + .route(Method::Get, "/healthz", |_, _| Response::ok().text("ok")) + .route(Method::Get, "/api/articles", |_, _| Response::ok().text("list")) + .route(Method::Get, "/api/articles/:id", |_, p| Response::ok().text(format!("get {}", p.get("id").unwrap()))) + .route(Method::Post, "/api/articles", |_, _| Response::created().text("create")); + + let resp = r.dispatch(&req(Method::Get, "/healthz")); + assert_eq!(resp.status.0, 200); + assert_eq!(resp.body, b"ok"); + + let resp = r.dispatch(&req(Method::Get, "/api/articles/42")); + assert_eq!(resp.body, b"get 42"); + + let resp = r.dispatch(&req(Method::Post, "/api/articles")); + assert_eq!(resp.status.0, 201); + } + + #[test] + fn returns_405_when_path_matches_but_method_does_not() { + let r = Router::new() + .route(Method::Get, "/api/articles", |_, _| Response::ok().text("list")); + let resp = r.dispatch(&req(Method::Post, "/api/articles")); + assert_eq!(resp.status.0, 405); + } + + #[test] + fn returns_404_when_no_path_matches() { + let r = Router::new() + .route(Method::Get, "/api/articles", |_, _| Response::ok().text("list")); + let resp = r.dispatch(&req(Method::Get, "/nope")); + assert_eq!(resp.status.0, 404); + } +} diff --git a/crates/rt/src/lib.rs b/crates/rt/src/lib.rs index 00bccad..0e59499 100644 --- a/crates/rt/src/lib.rs +++ b/crates/rt/src/lib.rs @@ -7,6 +7,7 @@ pub mod ast; pub mod compile; pub mod engine; +pub mod http; pub mod lexer; pub mod parser; pub mod runtime; diff --git a/crates/rt/src/server.rs b/crates/rt/src/server.rs index d6c7de7..885d564 100644 --- a/crates/rt/src/server.rs +++ b/crates/rt/src/server.rs @@ -1,4 +1,4 @@ -//! Axum REST server built from `service rest` blocks in the compiled catalog. +//! REST routing built from `service rest` blocks in the compiled catalog. //! //! For each type that declares `service rest "/path" expose ...`, we bind the //! exposed operations at the given path: @@ -9,61 +9,44 @@ //! create POST /path //! update PATCH /path/:id //! delete DELETE /path/:id -//! subscribe stubbed in Stage 2; wires up in Stage 3 -//! me GET /path/me (stubbed 501) +//! subscribe stubbed in Stage 2; wires up in Stage 3 (501) +//! me GET /path/me (501) //! ``` //! -//! The engine lives behind a single `Arc>` — matches the -//! single-threaded event-loop design of the runtime (see Phase 2 Concurrency -//! Model). +//! Phase-04 cutover: the axum + tokio backend was replaced with the +//! hand-rolled [`http::Router`](crate::http::Router) running on the +//! phase-02 [`EventLoop`](crate::runtime::EventLoop). Handlers are +//! synchronous, the engine sits behind `Arc>`, and +//! the binary owns one event loop on one thread. use crate::ast::{Operation, ServiceKind}; +use crate::compile::Catalog; use crate::engine::Engine; +use crate::http::{Method, Request, Response, RouteParams, Router, Status}; -use axum::{ - extract::{Path, State}, - http::StatusCode, - response::IntoResponse, - routing::{get, post}, - Json, Router, -}; use serde_json::{json, Value}; -use std::sync::Arc; -use tokio::sync::Mutex; +use std::sync::{Arc, Mutex}; pub type Shared = Arc>; -/// Per-route type context carried via axum `State`. -#[derive(Clone)] -struct TypeState { - engine: Shared, - ty: Arc, -} - -/// Build the fully-wired axum `Router` for a running engine. -/// The caller keeps the `Shared` handle alongside for Stage 3 subscription wiring. -pub fn router(engine: Shared, catalog: &crate::compile::Catalog) -> Router { - let mut app = Router::new() - .route("/", get(root)) - .route("/healthz", get(|| async { "ok" })); +/// Build the fully-wired [`Router`] for a running engine. +pub fn router(engine: Shared, catalog: &Catalog) -> Router { + let mut r = Router::new() + .route(Method::Get, "/", |_, _| root_response()) + .route(Method::Get, "/healthz", |_, _| Response::ok().text("ok")); for name in &catalog.order { let t = catalog.get(name).expect("type present"); for svc in &t.services { if svc.kind != ServiceKind::Rest { continue; } - let state = TypeState { - engine: engine.clone(), - ty: Arc::new(t.name.clone()), - }; - app = attach_rest(app, state, svc.path.clone(), &svc.expose); + r = attach_rest(r, engine.clone(), t.name.clone(), svc.path.clone(), &svc.expose); } } - - app + r } -async fn root() -> impl IntoResponse { - Json(json!({ +fn root_response() -> Response { + Response::ok().json(&json!({ "runtime": "wo", "stage": 2, "notes": "REST CRUD for each `service rest` block. /healthz for liveness. LIVE subscribe in Stage 3." @@ -71,122 +54,136 @@ async fn root() -> impl IntoResponse { } fn attach_rest( - mut app: Router, - state: TypeState, - path: String, - ops: &[Operation], + mut r: Router, + engine: Shared, + ty: String, + path: String, + ops: &[Operation], ) -> Router { - // Build a per-type sub-router with shared state, then merge. - let mut sub = Router::new(); + let id_path = format!("{path}/:id"); - // Collect collection-path handlers and id-path handlers separately so - // axum's `method_routing` merges correctly. - let mut collection = None::>; - let mut by_id = None::>; + // Register literal sub-paths (`/live`, `/me`) BEFORE the `/:id` param + // route — the first matching pattern wins, so `:id` would otherwise + // swallow "live" / "me" and produce a 400 invalid-id response. + for op in ops { + match op { + Operation::Subscribe => { + r = r.route(Method::Get, &format!("{path}/live"), + |_, _| Response::status(Status::NOT_IMPLEMENTED).text("LIVE subscriptions arrive in Stage 3")); + } + Operation::Me => { + r = r.route(Method::Get, &format!("{path}/me"), + |_, _| Response::status(Status::NOT_IMPLEMENTED).text("session layer not yet implemented")); + } + _ => {} + } + } for op in ops { match op { Operation::List => { - collection = Some(match collection.take() { - Some(r) => r.get(list_h), - None => get(list_h), - }); + let eng = engine.clone(); let ty = ty.clone(); + r = r.route(Method::Get, &path, move |req, params| list_h(&eng, &ty, req, params)); } Operation::Create => { - collection = Some(match collection.take() { - Some(r) => r.post(create_h), - None => post(create_h), - }); + let eng = engine.clone(); let ty = ty.clone(); + r = r.route(Method::Post, &path, move |req, params| create_h(&eng, &ty, req, params)); } Operation::Get => { - by_id = Some(match by_id.take() { - Some(r) => r.get(get_h), - None => get(get_h), - }); + let eng = engine.clone(); let ty = ty.clone(); + r = r.route(Method::Get, &id_path, move |req, params| get_h(&eng, &ty, req, params)); } Operation::Update => { - by_id = Some(match by_id.take() { - Some(r) => r.patch(update_h), - None => axum::routing::patch(update_h), - }); + let eng = engine.clone(); let ty = ty.clone(); + r = r.route(Method::Patch, &id_path, move |req, params| update_h(&eng, &ty, req, params)); } Operation::Delete => { - by_id = Some(match by_id.take() { - Some(r) => r.delete(delete_h), - None => axum::routing::delete(delete_h), - }); + let eng = engine.clone(); let ty = ty.clone(); + r = r.route(Method::Delete, &id_path, move |req, params| delete_h(&eng, &ty, req, params)); } - Operation::Subscribe => { - let p = format!("{path}/live"); - sub = sub.route(&p, get(subscribe_stub)); - } - Operation::Me => { - let p = format!("{path}/me"); - sub = sub.route(&p, get(me_stub)); - } - Operation::Custom => { /* reserved */ } + Operation::Subscribe | Operation::Me | Operation::Custom => {} } } - - if let Some(r) = collection { sub = sub.route(&path, r); } - if let Some(r) = by_id { sub = sub.route(&format!("{path}/:id"), r); } - - let sub = sub.with_state(state); - app = app.merge(sub); - app + r } // --- handlers --- -async fn list_h(State(st): State) -> impl IntoResponse { - let eng = st.engine.lock().await; - match eng.list(&st.ty) { - Ok(rows) => (StatusCode::OK, Json(json!(rows))).into_response(), - Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), +fn list_h(engine: &Shared, ty: &str, _req: &Request, _params: &RouteParams) -> Response { + let eng = engine.lock().unwrap(); + match eng.list(ty) { + Ok(rows) => Response::ok().json(&json!(rows)), + Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()), } } -async fn get_h(State(st): State, Path(id): Path) -> impl IntoResponse { - let eng = st.engine.lock().await; - match eng.get(&st.ty, id) { - Ok(Some(row)) => (StatusCode::OK, Json(json!(row))).into_response(), - Ok(None) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(), - Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), +fn get_h(engine: &Shared, ty: &str, _req: &Request, params: &RouteParams) -> Response { + let id = match parse_id(params) { + Ok(id) => id, + Err(r) => return r, + }; + let eng = engine.lock().unwrap(); + match eng.get(ty, id) { + Ok(Some(row)) => Response::ok().json(&json!(row)), + Ok(None) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")), + Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()), } } -async fn create_h(State(st): State, Json(body): Json) -> impl IntoResponse { - let mut eng = st.engine.lock().await; - match eng.create(&st.ty, body) { - Ok(row) => (StatusCode::CREATED, Json(json!(row))).into_response(), - Err(e) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(), +fn create_h(engine: &Shared, ty: &str, req: &Request, _params: &RouteParams) -> Response { + let body = match parse_json_body(req) { + Ok(v) => v, + Err(r) => return r, + }; + let mut eng = engine.lock().unwrap(); + match eng.create(ty, body) { + Ok(row) => Response::status(Status::CREATED).json(&json!(row)), + Err(e) => Response::status(Status::BAD_REQUEST).text(e.to_string()), } } -async fn update_h(State(st): State, Path(id): Path, Json(body): Json) -> impl IntoResponse { - let mut eng = st.engine.lock().await; - match eng.update(&st.ty, id, body) { - Ok(Some(row)) => (StatusCode::OK, Json(json!(row))).into_response(), - Ok(None) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(), - Err(e) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(), +fn update_h(engine: &Shared, ty: &str, req: &Request, params: &RouteParams) -> Response { + let id = match parse_id(params) { + Ok(id) => id, + Err(r) => return r, + }; + let body = match parse_json_body(req) { + Ok(v) => v, + Err(r) => return r, + }; + let mut eng = engine.lock().unwrap(); + match eng.update(ty, id, body) { + Ok(Some(row)) => Response::ok().json(&json!(row)), + Ok(None) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")), + Err(e) => Response::status(Status::BAD_REQUEST).text(e.to_string()), } } -async fn delete_h(State(st): State, Path(id): Path) -> impl IntoResponse { - let mut eng = st.engine.lock().await; - match eng.delete(&st.ty, id) { - Ok(true) => StatusCode::NO_CONTENT.into_response(), - Ok(false) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(), - Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), +fn delete_h(engine: &Shared, ty: &str, _req: &Request, params: &RouteParams) -> Response { + let id = match parse_id(params) { + Ok(id) => id, + Err(r) => return r, + }; + let mut eng = engine.lock().unwrap(); + match eng.delete(ty, id) { + Ok(true) => Response::no_content(), + Ok(false) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")), + Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()), } } -async fn subscribe_stub() -> impl IntoResponse { - (StatusCode::NOT_IMPLEMENTED, "LIVE subscriptions arrive in Stage 3") +fn parse_id(params: &RouteParams) -> Result { + params.get("id") + .and_then(|s| s.parse::().ok()) + .ok_or_else(|| Response::status(Status::BAD_REQUEST).text("invalid id")) } -async fn me_stub() -> impl IntoResponse { - (StatusCode::NOT_IMPLEMENTED, "session layer not yet implemented") +fn parse_json_body(req: &Request) -> Result { + if req.body.is_empty() { + return Ok(Value::Object(Default::default())); + } + serde_json::from_slice::(&req.body) + .map_err(|e| Response::status(Status::BAD_REQUEST).text(format!("invalid JSON: {e}"))) } /// Format the endpoint banner the CLI prints on startup. @@ -200,13 +197,13 @@ pub fn describe_routes(engine: &Engine) -> String { if svc.kind != ServiceKind::Rest { continue; } for op in &svc.expose { let (m, p, n) = match op { - Operation::List => ("GET ", svc.path.clone(), format!("list {name}")), - Operation::Get => ("GET ", format!("{}/:id", svc.path), format!("get {name}")), - Operation::Create => ("POST ", svc.path.clone(), format!("create {name}")), - Operation::Update => ("PATCH ", format!("{}/:id", svc.path), format!("update {name}")), - Operation::Delete => ("DELETE", format!("{}/:id", svc.path), format!("delete {name}")), - Operation::Subscribe => ("WS ", format!("{}/live", svc.path), format!("subscribe {name} (Stage 3)")), - Operation::Me => ("GET ", format!("{}/me", svc.path), "(Stage 3)".into()), + Operation::List => ("GET ", svc.path.clone(), format!("list {name}")), + Operation::Get => ("GET ", format!("{}/:id", svc.path), format!("get {name}")), + Operation::Create => ("POST ", svc.path.clone(), format!("create {name}")), + Operation::Update => ("PATCH ", format!("{}/:id", svc.path), format!("update {name}")), + Operation::Delete => ("DELETE", format!("{}/:id", svc.path), format!("delete {name}")), + Operation::Subscribe => ("WS ", format!("{}/live", svc.path), format!("subscribe {name} (Stage 3)")), + Operation::Me => ("GET ", format!("{}/me", svc.path), "(Stage 3)".into()), Operation::Custom => continue, }; out.push_str(&format!(" {m} {p:<30} {n}\n")); @@ -215,3 +212,79 @@ pub fn describe_routes(engine: &Engine) -> String { } out } + +#[cfg(test)] +mod tests { + use super::*; + use crate::compile::Catalog; + use crate::parser::parse; + + fn build(src: &str) -> (Shared, Router) { + let cat = Catalog::from_schemas(vec![parse(src).unwrap()]).unwrap(); + let eng = Arc::new(Mutex::new(Engine::new(cat.clone()))); + let r = router(eng.clone(), &cat); + (eng, r) + } + + fn req(method: Method, path: &str, body: &[u8]) -> Request { + Request { + method, + path: path.into(), + query: None, + headers: Default::default(), + body: body.to_vec(), + } + } + + #[test] + fn router_serves_crud_for_a_service_rest_block() { + let (_eng, r) = build(r#" +type Article { id: Id + title: Text + service rest "/api/articles" expose list, get, create, update, delete } +"#); + + // empty list + let resp = r.dispatch(&req(Method::Get, "/api/articles", b"")); + assert_eq!(resp.status.0, 200); + assert_eq!(resp.body, b"[]"); + + // create + let resp = r.dispatch(&req(Method::Post, "/api/articles", br#"{"title":"hello"}"#)); + assert_eq!(resp.status.0, 201); + + // get by id + let resp = r.dispatch(&req(Method::Get, "/api/articles/1", b"")); + assert_eq!(resp.status.0, 200); + + // delete + let resp = r.dispatch(&req(Method::Delete, "/api/articles/1", b"")); + assert_eq!(resp.status.0, 204); + + // 404 after delete + let resp = r.dispatch(&req(Method::Get, "/api/articles/1", b"")); + assert_eq!(resp.status.0, 404); + } + + #[test] + fn unexposed_method_yields_405() { + let (_eng, r) = build(r#" +type Tag { id: Id + label: Text + service rest "/api/tags" expose list, get } +"#); + let resp = r.dispatch(&req(Method::Post, "/api/tags", br#"{"label":"rust"}"#)); + assert_eq!(resp.status.0, 405); + } + + #[test] + fn live_subscribe_is_501() { + let (_eng, r) = build(r#" +type Article { id: Id + title: Text + service rest "/api/articles" expose list, subscribe } +"#); + let resp = r.dispatch(&req(Method::Get, "/api/articles/live", b"")); + assert_eq!(resp.status.0, 501); + } +} diff --git a/docs/plan/03-hand-rolled-http.md b/docs/plan/done/03-hand-rolled-http.md similarity index 100% rename from docs/plan/03-hand-rolled-http.md rename to docs/plan/done/03-hand-rolled-http.md