feat(serve): supervised listener pool (v0.2 chunk 1)

OneForOne supervisor on the root actor, Restart::Permanent per listener.
ChildSpec factory owns its fd via Arc<OwnedFd> (OwnedFd gains Sync);
restarts reuse the fd — no re-dup, no fd-less window. wait_readable
failure now exits-and-restarts instead of being fatal. Conn actors stay
unsupervised bare spawns. Test hook INJECT_LISTENER_PANICS panics before
accept so a pending connection must survive into the restarted listener.

Roadmap: chunk 1 marked done; chunk 3 fork resolved to smarm-native
timed waits (RFC 008 landed at 393cdd0, reaper option removed); chunk 2
drain-then-stop decided; v0.4-3(b) unblocked.

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
+  one waiter per RawFd (`AlreadyExists`, now surfaced as `Err` via the
-  direction per parked actor. Concurrent read+write on one connection needs
+  `try_select` twins) and one direction per parked actor. RFC 008 phase 1
-  either the dup-fd trick (serve.rs already uses it for the listener pool)
+  (landed `393cdd0`) gives first-of(fd, channel) via fd arms in select —
-  or a new smarm primitive. Decide in v0.2 design, before the frame codec.
+  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
+### 1. Supervised listener pool ✅
-Replace bare spawns in `serve_with` with a supervisor:
+Landed: `serve_with` runs a `Strategy::OneForOne` supervisor on the root
-`Strategy::OneForOne`, one `ChildSpec` per dup'd listener fd,
+actor, one `ChildSpec` per dup'd listener fd, `Restart::Permanent`. A
-`Restart::Permanent`. A listener panic restarts that listener; the pool
+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
+Open question resolved as leaned: the ChildSpec factory owns the fd via
-own its fd across restarts via the ChildSpec closure? Leaning: closure
+`Arc<OwnedFd>` (`OwnedFd` gained `Sync` for this) and every (re)start
-captures the `OwnedFd`, re-registers on restart. No re-dup needed.
+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
+2. **Decided: drain-then-stop with a deadline** — in-flight requests get
-   immediate `request_stop` (unwind closes fds cleanly via `OwnedFd::drop`).
+   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
+actor monitors-free self-deregisters via a drop guard cast. Pid set only
-is also chunk 3's reaper and the future ws/channels introspection point,
+(no activity stamps — chunk 3 went per-conn timed waits, no reaper); still
-so build it once here.
+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
+`read_some` parks in `wait_readable_timeout` instead of `wait_readable`;
-  `last_activity`; a ticker (`smarm::sleep` loop) sweeps and
+`Ok(false)` means the deadline passed and the conn actor closes. Per-conn
-  `request_stop`s expired conns. Works today, zero smarm changes.
+semantics, no activity stamping, no sweep ticker. Semantics note: a timed
-  Granularity = sweep interval; conns must stamp activity (one cast per
+read wait enforces *idle-between-reads* — `keep_alive_timeout` between
-  request — measurable but small).
+requests, and slowloris dies because each stalled read has a deadline.
-- **(b) `wait_readable_timeout` in smarm.** A first-of(fd, timer) wait.
+`request_timeout` as a whole-request wall clock is NOT what this gives;
-  The v0.7 epoch-stamped consuming-unpark machinery exists precisely for
+decide in implementation whether to track a per-request deadline across
-  multi-arm waits (`select_timeout` is the channel-flavoured proof), but
+reads (cheap: one `Instant` in the conn loop, pass the min of both
-  fd arms aren't `Selectable` today — this is real smarm work in
+budgets to each wait) or redefine `request_timeout` as per-read. Leaning
-  `wait_fd`/io thread. Cleaner per-conn semantics, benefits every future
+the `Instant`: it keeps the advertised config honest.
-  smarm io consumer, no reaper bookkeeping.
 
-Leaning **(a) now, (b) later**: (a) ships the security fix this cycle and
+urus is the first real consumer of the timed wrappers — note the RFC 008
-the registry exists anyway; (b) becomes a smarm roadmap item whose landing
+reviewer flag (`wait_fd` wraps in `NoPreempt`, select arms don't); the
-lets the reaper's stamps go away. But if you'd rather not build throwaway:
+crud example under load doubles as that confirmation.
-(b) first is defensible — say the word.
 
 ### 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
+   via comment lines on a timer. An SSE conn is just a Stream body with
-   the reaper exempts from `request_timeout` (keep-alive ping instead).
+   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<Frame>`; 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
+   - **(b) One actor, smarm first-of(fd-readable, channel)**: fd arms in
-     arms in select — the same smarm work as v0.2-3(b). Simpler topology,
+     select landed with RFC 008 (`393cdd0`) — no longer blocked. Simpler
-     no dup, but blocked on smarm.
+     topology, no dup; use the fallible `try_select` twins (registration
-   Default (a); revisit if (b) lands for timeouts anyway.
+     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,

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<OwnedFd>`.
+unsafe impl Sync for OwnedFd {}
+
 // ---------------------------------------------------------------------------
 // bind_and_listen
 // ---------------------------------------------------------------------------

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<OwnedFd> {
 // 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<OwnedFd>, 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
+// `Config::default()`) and runs a one-for-one supervisor over the listener
-// graceful-shutdown signal in v1; the runtime exits when all actors exit,
+// pool on the root actor. A panicking listener is restarted on the same
-// which they don't (listener loops are infinite). Ctrl-C is your friend.
+// (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.
+    // One connection-actor-spawning loop per listener pool slot. Each gets
-    // Each gets its own dup'd fd so epoll registrations don't collide.
+    // 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 sup = OneForOne::new().strategy(Strategy::OneForOne);
-        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 || {
+            sup = sup.child(ChildSpec::new(Restart::Permanent, move || {
                 println!("urus: listener {} starting", i);
-                listener_loop(lfd, p, limits);
+                listener_loop(lfd.clone(), p.clone(), 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();
         }
+        // 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(())

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);
+    }
+}