diff --git a/ROADMAP.md b/ROADMAP.md index 65303b5..4f9f3ef 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -7,7 +7,8 @@ ## State assessment (2026-06-11) -Verified against smarm master (`v0.8`, slot slab + park-epoch + select): +Verified against smarm master (`393cdd0`: v0.8 slot slab + park-epoch + +select, plus RFC 008 phase 1 fd arms / timed fd waits): - **Builds clean, 24/24 tests pass.** The scheduler surface urus consumes (`spawn`, `wait_readable`, `wait_writable`, `sleep`, `init`, `Config`, @@ -27,10 +28,11 @@ Verified against smarm master (`v0.8`, slot slab + park-epoch + select): - **No streaming response path.** `RespBody` is `Empty | Bytes`. Blocks SSE (a spec v1 goal that slipped) and chunked responses. - **Full-duplex constraint (matters for WebSocket).** smarm's epoll model is - one waiter per RawFd (`AlreadyExists` on the second registration) and one - direction per parked actor. Concurrent read+write on one connection needs - either the dup-fd trick (serve.rs already uses it for the listener pool) - or a new smarm primitive. Decide in v0.2 design, before the frame codec. + one waiter per RawFd (`AlreadyExists`, now surfaced as `Err` via the + `try_select` twins) and one direction per parked actor. RFC 008 phase 1 + (landed `393cdd0`) gives first-of(fd, channel) via fd arms in select — + the duplex options in v0.4-3 are both open now. Decide in v0.4 design, + before the frame codec. --- @@ -39,51 +41,60 @@ Verified against smarm master (`v0.8`, slot slab + park-epoch + select): Adopt the primitives urus predates. No protocol surface changes; `Plug`/`Conn` untouched. Chunks are independently landable, in order: -### 1. Supervised listener pool -Replace bare spawns in `serve_with` with a supervisor: -`Strategy::OneForOne`, one `ChildSpec` per dup'd listener fd, -`Restart::Permanent`. A listener panic restarts that listener; the pool +### 1. Supervised listener pool ✅ +Landed: `serve_with` runs a `Strategy::OneForOne` supervisor on the root +actor, one `ChildSpec` per dup'd listener fd, `Restart::Permanent`. A +listener panic (or a transient `wait_readable` failure, which now exits +and restarts instead of being fatal) restarts that listener; the pool never silently shrinks. Connection actors stay unsupervised bare spawns (per spec: a connection is cheap and its failure is local — a 500 path, -not a restart path). +not a restart path); `spawn` parents them under the listener, which has no +supervisor channel, so their deaths are invisible to the pool supervisor +by construction. -Open question: should the restarted child re-dup from listener 0's fd, or -own its fd across restarts via the ChildSpec closure? Leaning: closure -captures the `OwnedFd`, re-registers on restart. No re-dup needed. +Open question resolved as leaned: the ChildSpec factory owns the fd via +`Arc` (`OwnedFd` gained `Sync` for this) and every (re)start +reuses it. No re-dup, no window where a pool slot has no fd. Covered by +`tests/integration.rs::panicking_listener_restarts` (pool of 1, panic +injected before `accept` so the pending connection must survive into the +restarted listener). ### 2. Graceful shutdown `serve_with` gains a shutdown signal (an `urus::Handle` with `.shutdown()`, backed by a channel + `request_stop` fan-out): 1. Stop listeners (no new accepts; listener fds close). -2. Stop connection actors — either drain-then-stop with a deadline, or - immediate `request_stop` (unwind closes fds cleanly via `OwnedFd::drop`). +2. **Decided: drain-then-stop with a deadline** — in-flight requests get + to finish; `request_stop` after the deadline (unwind closes fds cleanly + via `OwnedFd::drop`). Drain also keeps the door open for an + apache-reload-style restart later. 3. Supervisor winds down; `rt.run` returns; `serve_with` returns. Requires tracking live connection pids. Cheapest honest version: a `gen_server` connection registry — listener casts `{Started, pid}`, conn -actor monitors-free self-deregisters via a drop guard cast. This registry -is also chunk 3's reaper and the future ws/channels introspection point, -so build it once here. +actor monitors-free self-deregisters via a drop guard cast. Pid set only +(no activity stamps — chunk 3 went per-conn timed waits, no reaper); still +the future ws/channels introspection point, so build it once here. ### 3. Enforce the timeouts we advertise -**Decision fork — pick one before implementing:** +**Decided: smarm-native timed waits** (the fork's option (a), a reaper in +urus, died when RFC 008 landed at `393cdd0` — `wait_readable_timeout` / +`wait_writable_timeout` exist upstream now, so there is nothing for the +reaper to be the stopgap for): -- **(a) Reaper in urus.** The chunk-2 registry tracks per-conn - `last_activity`; a ticker (`smarm::sleep` loop) sweeps and - `request_stop`s expired conns. Works today, zero smarm changes. - Granularity = sweep interval; conns must stamp activity (one cast per - request — measurable but small). -- **(b) `wait_readable_timeout` in smarm.** A first-of(fd, timer) wait. - The v0.7 epoch-stamped consuming-unpark machinery exists precisely for - multi-arm waits (`select_timeout` is the channel-flavoured proof), but - fd arms aren't `Selectable` today — this is real smarm work in - `wait_fd`/io thread. Cleaner per-conn semantics, benefits every future - smarm io consumer, no reaper bookkeeping. +`read_some` parks in `wait_readable_timeout` instead of `wait_readable`; +`Ok(false)` means the deadline passed and the conn actor closes. Per-conn +semantics, no activity stamping, no sweep ticker. Semantics note: a timed +read wait enforces *idle-between-reads* — `keep_alive_timeout` between +requests, and slowloris dies because each stalled read has a deadline. +`request_timeout` as a whole-request wall clock is NOT what this gives; +decide in implementation whether to track a per-request deadline across +reads (cheap: one `Instant` in the conn loop, pass the min of both +budgets to each wait) or redefine `request_timeout` as per-read. Leaning +the `Instant`: it keeps the advertised config honest. -Leaning **(a) now, (b) later**: (a) ships the security fix this cycle and -the registry exists anyway; (b) becomes a smarm roadmap item whose landing -lets the reaper's stamps go away. But if you'd rather not build throwaway: -(b) first is defensible — say the word. +urus is the first real consumer of the timed wrappers — note the RFC 008 +reviewer flag (`wait_fd` wraps in `NoPreempt`, select arms don't); the +crud example under load doubles as that confirmation. ### 4. Hygiene - Registry names: `urus.server`, `urus.listener.{n}` via `smarm::register` @@ -111,8 +122,9 @@ is "a response that never ends"). cycle, the codec knowledge is fresh. 3. SSE sugar: `Conn::sse()` → an `EventSender` (`.send(event, data)`), sets `content-type: text/event-stream`, disables buffering, heartbeats - via comment lines on a timer. An SSE conn is just a Stream body that - the reaper exempts from `request_timeout` (keep-alive ping instead). + via comment lines on a timer. An SSE conn is just a Stream body with + the per-read deadline disabled (or set to the heartbeat interval) — + liveness comes from the keep-alive ping, not `request_timeout`. --- @@ -130,10 +142,15 @@ is "a response that never ends"). - **(a) Two actors, dup'd fd**: reader actor on the dup, writer actor on the original, writer owns a `Receiver`; the user handler talks to both via channels. Proven trick (listener pool), works today. - - **(b) One actor, smarm first-of(fd-readable, channel)**: needs fd - arms in select — the same smarm work as v0.2-3(b). Simpler topology, - no dup, but blocked on smarm. - Default (a); revisit if (b) lands for timeouts anyway. + - **(b) One actor, smarm first-of(fd-readable, channel)**: fd arms in + select landed with RFC 008 (`393cdd0`) — no longer blocked. Simpler + topology, no dup; use the fallible `try_select` twins (registration + errors, incl. `AlreadyExists`, surface as `Err` and the wait fully + retires). + Was "default (a), revisit if (b) lands" — (b) landed, so this is a + live choice for the v0.4 design pass. Note RFC 008 is phase 1: + single-waiter-per-fd stands, `dup` remains the documented answer for + true simultaneous read+write waits. 4. Handler API sketch (settle in design pass): trait with `on_message/on_close` + a `WsSender` handle — gen_server-shaped, so a ws connection *is* an actor you can monitor, link, and register. @@ -182,8 +199,6 @@ PubSub underneath. - **SSE ergonomics round 2** — auto-reconnect support (`Last-Event-ID`), event-id bookkeeping, backpressure policy on slow consumers. - **TLS** — still deferred; acceptor design still must not preclude it. -- **`wait_readable_timeout` / fd-in-select upstreaming** — the smarm-side - item that simplifies v0.2-3 and v0.4-3 retroactively. - **Middleware ecosystem** — compression, static files, auth plugs. - **Bench suite** — `urus-bench-spec.md` exists in the artefact store; wire it up once v0.2 lands (supervision changes the hot path not at all, diff --git a/src/net.rs b/src/net.rs index 701711c..4d34223 100644 --- a/src/net.rs +++ b/src/net.rs @@ -52,6 +52,13 @@ impl Drop for OwnedFd { // becomes the unique closer. unsafe impl Send for OwnedFd {} +// SAFETY: shared references only expose `as_raw(&self)` — reading an +// integer. No interior mutability; `into_raw` takes `self` by value and is +// unreachable through a shared reference. Needed so a supervisor +// `ChildSpec` factory (an `Fn` closure, hence shared across restarts) can +// own its listener fd via `Arc`. +unsafe impl Sync for OwnedFd {} + // --------------------------------------------------------------------------- // bind_and_listen // --------------------------------------------------------------------------- diff --git a/src/serve.rs b/src/serve.rs index 6baf7af..a722eea 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -16,9 +16,13 @@ use crate::conn_actor::{run_connection, ConnLimits}; use crate::net::{accept_nonblocking, bind_and_listen, OwnedFd}; use crate::plug::Pipeline; +use smarm::{ChildSpec, OneForOne, Restart, Strategy}; + use std::io::{self, ErrorKind}; use std::net::{SocketAddr, ToSocketAddrs}; use std::os::fd::RawFd; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; use std::time::Duration; // --------------------------------------------------------------------------- @@ -85,9 +89,24 @@ fn dup_fd(fd: RawFd) -> io::Result { // listener actor body // --------------------------------------------------------------------------- -fn listener_loop(listener: OwnedFd, pipeline: Pipeline, limits: ConnLimits) { +/// Test-only fault injection. When nonzero, the next accept-loop iteration +/// of whichever listener gets there first decrements this and panics — +/// *before* calling `accept`, so a pending connection stays in the kernel +/// backlog and must be picked up by the restarted listener. Cost when idle +/// is one relaxed load per accept-loop iteration (each of which already +/// pays a syscall). Not public API. +#[doc(hidden)] +pub static INJECT_LISTENER_PANICS: AtomicU32 = AtomicU32::new(0); + +fn listener_loop(listener: Arc, pipeline: Pipeline, limits: ConnLimits) { let fd = listener.as_raw(); loop { + if INJECT_LISTENER_PANICS + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_sub(1)) + .is_ok() + { + panic!("urus: injected listener panic (test hook)"); + } match accept_nonblocking(fd) { Ok(client) => { // Hand the fd off to a new connection actor. spawn() is @@ -101,7 +120,11 @@ fn listener_loop(listener: OwnedFd, pipeline: Pipeline, limits: ConnLimits) { // 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. + // epoll registration failed. Under supervision a + // `return` is a `Signal::Exit` and `Restart::Permanent` + // restarts us — a transient failure (e.g. EMFILE on + // the epoll set) now heals instead of silently + // shrinking the pool. eprintln!("urus: listener wait_readable failed: {we}"); return; } @@ -119,57 +142,67 @@ fn listener_loop(listener: OwnedFd, pipeline: Pipeline, limits: ConnLimits) { } } } - // listener OwnedFd drops here, closing the dup'd fd. + // The Arc clone we were started with drops here (normal exit or + // unwind), but the ChildSpec factory holds another — the fd outlives + // any one incarnation of this listener. } // --------------------------------------------------------------------------- -// serve_with — main entry. Boots smarm, spawns listeners, blocks. +// serve_with — main entry. Boots smarm, supervises 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. +// `Config::default()`) and runs a one-for-one supervisor over the listener +// pool on the root actor. A panicking listener is restarted on the same +// (still-open) fd instead of silently shrinking the accept pool. Connection +// actors stay unsupervised bare spawns — per spec, a connection is cheap +// and its failure is local: a 500 path, not a restart path. (`spawn` from a +// listener parents the conn actor under that listener, which never +// registers a supervisor channel, so conn deaths are invisible to the pool +// supervisor by construction.) +// +// No graceful-shutdown signal yet (v0.2 chunk 2); the supervisor loop never +// returns because listeners never reach a terminal state. 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. + // One connection-actor-spawning loop per listener pool slot. Each gets + // its own dup'd fd so epoll registrations don't collide. Each fd is + // owned by its ChildSpec's factory closure (via `Arc`): a restarted + // listener re-enters `accept`/`wait_readable` on the same fd — no + // re-dup, no window where the slot has no fd. let mut listener_fds = Vec::with_capacity(config.listener_pool); - listener_fds.push(listener); // primary keeps the original + listener_fds.push(Arc::new(listener)); // primary keeps the original for _ in 1..config.listener_pool { let dup = dup_fd(listener_fds[0].as_raw())?; - listener_fds.push(dup); + listener_fds.push(Arc::new(dup)); } - let limits = config.to_conn_limits(); + 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); + let mut sup = OneForOne::new().strategy(Strategy::OneForOne); for (i, lfd) in listener_fds.into_iter().enumerate() { let p = pipeline.clone(); - let h = smarm::spawn(move || { + sup = sup.child(ChildSpec::new(Restart::Permanent, 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(); + listener_loop(lfd.clone(), p.clone(), limits); + })); } + // Supervision loop on the root actor. Default intensity (3 per 5s) + // applies per the runtime; a listener crash-looping faster than + // that trips the cap and tears the pool down — loud failure over a + // zombie server. + sup.run(); }); Ok(()) diff --git a/tests/integration.rs b/tests/integration.rs index 56e8c51..e4b3936 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -225,3 +225,62 @@ fn middleware_can_short_circuit() { assert_eq!(http_status(&resp), 200); assert_eq!(http_body(&resp), b"ok"); } + +// --------------------------------------------------------------------------- +// Supervision (v0.2 chunk 1) +// --------------------------------------------------------------------------- + +/// A panicking listener must be restarted by the pool supervisor, and the +/// connection that was pending when it died must still be served. +/// +/// Pool size is 1 on purpose: with a sibling listener the test would pass +/// even if restarts were broken (the sibling would pick up the accept). +/// With one listener, the request after the injected panic can only +/// succeed if a fresh listener came up on the same fd. +#[test] +fn panicking_listener_restarts() { + let pipe = Pipeline::new().plug( + Router::new().get("/ping", |c: Conn, _n: Next| c.put_status(200).put_body("pong")) + ); + + let port = free_port(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + std::thread::spawn(move || { + let cfg = Config { + listener_pool: 1, + scheduler_threads: Some(2), + ..Config::new(addr) + }; + serve_with(cfg, pipe).unwrap(); + }); + for _ in 0..50 { + if TcpStream::connect(addr).is_ok() { + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + + // Sanity: server answers before the fault. + let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 200); + + // Arm the fault: the listener's next accept-loop iteration panics + // *before* accepting, so our connection waits in the kernel backlog + // until the restarted listener picks it up. + urus::serve::INJECT_LISTENER_PANICS.store(1, std::sync::atomic::Ordering::Relaxed); + + let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 200, "request pending across the panic was not served"); + assert_eq!(http_body(&resp), b"pong"); + assert_eq!( + urus::serve::INJECT_LISTENER_PANICS.load(std::sync::atomic::Ordering::Relaxed), + 0, + "fault was never consumed — listener didn't wake for the connection" + ); + + // The restarted listener keeps serving. + for _ in 0..5 { + let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); + assert_eq!(http_status(&resp), 200); + } +}