Markk116/urus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs+examples: v0.3 endpoint API; crud converted to the app-owns-the-tree shape

Browse Source 
- crud is now the demonstrator: app owns smarm runtime + root supervisor,
  store actor and urus::endpoint as ordered siblings (store first, so
  reverse-order shutdown drains HTTP before stopping the store),
  Shutdown::Infinity on the endpoint, shutdown via
  rt.handle().request_shutdown(root_sup) from the stdin thread.
  Deletes two kludges the old shape forced:
    * static OnceLock<Sender> spawn-on-first-use -> a supervised child
      that self-registers a typed Name; handlers use smarm::send per
      request and turn 'between incarnations' into a 503 instead of
      panicking on a dropped store.
    * static SHUTTING_DOWN AtomicBool + 250ms recv_timeout poll in the
      store loop and in the SSE ticker -> a plain park; the tree stops
      both. Smoke-tested live: CRUD round-trips, SSE stream, clean drain
      with the stream open, port closed after.
- Other examples stay short and on serve*, updated for the split config
  (serve_with(cfg, smarm::Config, pipe) / serve_with_shutdown(..., signal)).
  plain_serve's URUS_SCHED_THREADS now builds a smarm::Config.
  ws_chat's doc block explains the OnceLock is a serve*-only workaround
  and points at crud for the clean shape.
- README: new 'Your Own Supervision Tree' section (endpoint as the real
  API), graceful shutdown reframed as the serve*-only path, Config table
  loses scheduler_threads and gains name, PubSub rule 1 notes the
  supervised-sibling alternative.

111 lib + 50 integration + 2 doc tests green; clippy clean.
This commit is contained in:
Claude
2026-08-20 13:20:28 +00:00
parent 099b6bc320
commit b37888ec2c
9 changed files with 181 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/crud.rs
+88 -68
View File
@@ -1,11 +1,18 @@
//! CRUD example: a tiny user database with JSON persistence.
//!
//! Demonstrates urus and the actor model together:
//! - The pipeline is shared (Arc) across all connection actors.
//! - Handlers do NOT take a lock or share mutable state directly.
//! - A single "store" actor owns the data; handlers send it a request
//! via a channel and block on the reply. Serialization is structural —
//! the store processes one request at a time, no Mutex needed.
//! Demonstrates urus and the actor model together, in the shape a real
//! application should use (v0.3):
//! - The APP owns the smarm runtime and the root supervisor. urus is one
//! child in that tree — `urus::endpoint(...)` — and the store actor is
//! an ordered sibling started BEFORE it, so the supervisor's
//! reverse-order shutdown drains HTTP first and only then stops the
//! store. No handler can be mid-request against a store that is gone.
//! - Handlers do NOT take a lock or share mutable state directly. A
//! single "store" actor owns the data; handlers address it by
//! registered name and block on a reply channel. Serialization is
//! structural — one request at a time, no Mutex.
//! - Both actors are supervised: kill the store (or let it panic) and it
//! restarts from the JSON file, with the endpoint left alone.
//! - On every mutating request the store writes the JSON file. Read
//! requests don't touch disk.
//!
@@ -23,9 +30,8 @@
//! curl -s http://localhost:8080/users/1
use serde::{Deserialize, Serialize};
use smarm::{channel, Sender};
use std::sync::OnceLock;
use urus::{serve_with_shutdown, shutdown_handle, Config, Conn, Next, Pipeline, Router};
use smarm::{channel, ChildSpec, Name, OneForOne, Restart, Sender};
use urus::{Config, Conn, Next, Pipeline, Router};
// ---------------------------------------------------------------------------
// Domain
@@ -66,7 +72,12 @@ const DB_PATH: &str = "/tmp/urus-crud.json";
// Store actor body
// ---------------------------------------------------------------------------
fn store_loop(rx: smarm::Receiver<Request>) {
fn store_loop() {
let (tx, rx) = channel::<Request>();
// Self-registration: the name is bound before the first recv, and it
// is re-bound automatically on every restart.
smarm::register(STORE, tx).expect("crud.store name already taken");
// Load on start. Missing file = empty store. Corrupt file = panic; we
// don't auto-rebuild because silently losing data is worse than failing
// loud.
@@ -78,19 +89,10 @@ fn store_loop(rx: smarm::Receiver<Request>) {
let mut next_id: u64 = users.iter().map(|u| u.id).max().unwrap_or(0) + 1;
loop {
// recv with a timeout rather than a bare recv: the store must be
// stoppable at shutdown, but a cross-thread Sender::send (from the
// stdin thread) can't wake a parked actor — same smarm limitation
// that motivates urus's SHUTDOWN_POLL. So we wake on our own timer
// and poll the flag. This poll dies with that limitation too.
let req = match rx.recv_timeout(std::time::Duration::from_millis(250)) {
// A plain park. The supervisor stops this actor at shutdown (after
// the endpoint has drained), so there is nothing to poll for.
let req = match rx.recv() {
Ok(r) => r,
Err(smarm::channel::RecvTimeoutError::Timeout) => {
if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
return;
}
continue;
}
Err(_) => return, // all senders dropped
};
match req {
@@ -169,25 +171,29 @@ fn persist(users: &[User]) {
// Handler helpers
// ---------------------------------------------------------------------------
//
// Once-cell trick: the store actor is spawned the first time a handler
// runs (smarm requires `spawn` to be called from inside an actor — which
// connection actors are). After that all handlers share the same Sender.
// Simpler than threading the Sender through the pipeline at startup.
// The store is a supervised child that registers its own inbox under a
// typed name; handlers resolve it per send. That replaces the old
// `OnceLock<Sender>` spawn-on-first-use trick — which had no supervisor,
// no restart, and no defined shutdown point — with a plain actor whose
// lifecycle the tree owns. A restart re-registers the same name, so
// in-flight handlers heal on their next send.
static STORE_TX: OnceLock<Sender<Request>> = OnceLock::new();
const STORE: Name<Request> = Name::new("crud.store");
// Set by the stdin thread at shutdown; the store actor polls it (see
// store_loop). An always-on app actor that never returns would otherwise
// block smarm's AllDone and keep serve_with_shutdown from returning.
static SHUTTING_DOWN: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
/// Send to the store and wait for its reply. `Err` only if the store is
/// between incarnations (restarting); the handler turns that into a 503
/// rather than pretending.
fn ask(make: impl FnOnce(Sender<(u16, Vec<u8>)>) -> Request) -> Option<(u16, Vec<u8>)> {
let (tx, rx) = channel::<(u16, Vec<u8>)>();
smarm::send(STORE, make(tx)).ok()?;
rx.recv().ok()
}
fn store() -> &'static Sender<Request> {
STORE_TX.get_or_init(|| {
let (tx, rx) = channel::<Request>();
smarm::spawn(move || store_loop(rx));
tx
})
fn reply(conn: Conn, r: Option<(u16, Vec<u8>)>) -> Conn {
match r {
Some((status, body)) => json(conn, status, body),
None => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
}
}
fn json(conn: Conn, status: u16, body: Vec<u8>) -> Conn {
@@ -206,16 +212,13 @@ fn parse_id(s: &str) -> Option<u64> {
/// SSE demo: `curl -N localhost:8080/ticker` streams a tick every second
/// (with `: keep-alive` comments if it ever goes quiet). The producer
/// exits on SseClosed (client gone / write timeout / shutdown drain) or
/// when the example is shutting down.
/// exits on SseClosed — client gone, write timeout, or the drain stopping
/// its connection actor. Nothing to flag: the tree's shutdown reaches it.
fn ticker(conn: Conn, _next: Next) -> Conn {
let (conn, events) = conn.sse();
smarm::spawn(move || {
let mut n: u64 = 0;
loop {
if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
return; // dropping `events` ends the stream cleanly
}
if events.send("tick", &n.to_string()).is_err() {
return; // SseClosed
}
@@ -227,18 +230,14 @@ fn ticker(conn: Conn, _next: Next) -> Conn {
}
fn list(conn: Conn, _next: Next) -> Conn {
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::List { reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
let r = ask(|reply| Request::List { reply });
reply(conn, r)
}
fn create(conn: Conn, _next: Next) -> Conn {
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Create { body, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
let r = ask(|reply| Request::Create { body, reply });
reply(conn, r)
}
fn get_one(conn: Conn, _next: Next) -> Conn {
@@ -246,10 +245,8 @@ fn get_one(conn: Conn, _next: Next) -> Conn {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Get { id, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
let r = ask(|reply| Request::Get { id, reply });
reply(conn, r)
}
fn update(conn: Conn, _next: Next) -> Conn {
@@ -258,10 +255,8 @@ fn update(conn: Conn, _next: Next) -> Conn {
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Update { id, body, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
let r = ask(|reply| Request::Update { id, body, reply });
reply(conn, r)
}
fn delete(conn: Conn, _next: Next) -> Conn {
@@ -269,10 +264,8 @@ fn delete(conn: Conn, _next: Next) -> Conn {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Delete { id, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
let r = ask(|reply| Request::Delete { id, reply });
reply(conn, r)
}
// ---------------------------------------------------------------------------
@@ -305,20 +298,47 @@ fn main() {
);
let cfg = Config::new("127.0.0.1:8080".parse().unwrap());
// Bind here, on this thread: an address-in-use error is a startup
// failure, not an actor crash. The fds outlive any restart of the
// endpoint child.
let endpoint = urus::endpoint(cfg, pipeline).expect("bind 127.0.0.1:8080");
println!("urus-crud: DB at {DB_PATH}");
println!("urus-crud: listening on 127.0.0.1:8080 — press Enter to shut down");
let rt = smarm::init(smarm::Config::default());
let handle = rt.handle();
let (sup_tx, sup_rx) = std::sync::mpsc::channel();
// Graceful shutdown on stdin-Enter: a plain OS thread blocks on
// read_line and fires the handle. No signal handling crate needed.
let (handle, signal) = shutdown_handle();
// read_line and shuts the ROOT SUPERVISOR down. No signal-handling
// crate needed, and no urus-specific shutdown plumbing — this is
// exactly what a SIGTERM handler would do.
std::thread::spawn(move || {
let sup: smarm::Pid = sup_rx.recv().expect("supervisor pid");
let mut line = String::new();
let _ = std::io::stdin().read_line(&mut line);
println!("urus-crud: shutting down (draining in-flight requests)…");
SHUTTING_DOWN.store(true, std::sync::atomic::Ordering::Relaxed);
handle.shutdown();
handle.request_shutdown(sup);
});
serve_with_shutdown(cfg, pipeline, signal).unwrap();
rt.run(move || {
let sup = smarm::spawn(move || {
OneForOne::new()
// Store FIRST: reverse-order shutdown therefore stops it
// LAST, after the endpoint has finished draining.
.child(ChildSpec::new(Restart::Permanent, store_loop))
.child(
ChildSpec::new(Restart::Permanent, endpoint)
// The endpoint bounds its own drain with
// Config::drain_timeout; a shorter supervisor
// deadline would cut that drain in half.
.shutdown(smarm::supervisor::Shutdown::Infinity),
)
.run()
});
let _ = sup_tx.send(sup.pid());
let _ = sup.join();
});
println!("urus-crud: bye");
}