diff --git a/README.md b/README.md index da8471a..8c53e38 100644 --- a/README.md +++ b/README.md @@ -299,6 +299,50 @@ client is caught when a write stalls past `write_timeout`, and graceful shutdown force-stops the connection at the drain deadline. See `examples/ws_echo.rs`. +The `WsHandler` trait also has a defaulted `on_open(&mut self, sender)`, +called once before the frame loop — the place to subscribe or spawn +producers for clients that may never send. Note the client sees the 101 +*before* `on_open` runs; a client that must know its subscriptions are +live uses an application-level ack (any reply proves `on_open` completed, +callbacks being sequential). + +## Pub/Sub (v0.5) + +`urus::pubsub` is a local-node, phoenix_pubsub-shaped topic broadcaster — +independent of HTTP (it imports only smarm) and built for the WebSocket +relay pattern: + +```rust +let bus: PubSub = PubSub::new(); // in-runtime only! +let rx = bus.subscribe("room:lobby")?; // Receiver> +bus.broadcast("room:lobby", "hi".to_string())?; +bus.broadcast_from(smarm::self_pid(), "room:lobby", "no echo".into())?; +``` + +One generic instance per message domain; payloads broadcast as `Arc` +(one allocation per broadcast). `subscribe` is idempotent per +`(pid, topic)` and pins the subscription to the **calling actor** — its +death prunes the entry via a monitor; a dropped `Receiver` is pruned +lazily on the next broadcast. `subscribe_as(pid, topic)` pins to an +explicit pid for relay patterns. Mailboxes are unbounded: `broadcast` +never blocks the table, and a slow subscriber's memory bill is bounded +by the two cleanup paths above. + +Two composition rules that matter (both enforced by +`shutdown_with_open_chat_terminates` in the integration suite): + +1. `PubSub::new()` spawns an actor, so it must run in-runtime — build it + lazily from a handler via a **non-static** `Arc>>` + captured by the route closure. A `static` cell pins the table actor + forever and graceful shutdown never returns. +2. Relay/producer actors hold the `Receiver` (plus e.g. a `WsSender` + clone) — **never a `PubSub` clone**, or relay and table keep each + other alive past shutdown. + +See [`examples/ws_chat.rs`](examples/ws_chat.rs): rooms as topics, +`on_open` subscribes + spawns the relay, `on_message` uses +`broadcast_from` so the speaker isn't echoed. + ## Examples ### CRUD with Actor Ownership diff --git a/ROADMAP.md b/ROADMAP.md index c82262e..d8f8c50 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -216,23 +216,52 @@ force-stopped at the drain deadline on shutdown; full suite green. 4. **Handler API** ✅ — folded into chunk 3 above (was: settle in design pass; settled: gen_server-shaped trait + WsSender handle). -## v0.5 — PubSub +## v0.5 — PubSub ✅ (landed 2026-06, chunks 1-2) `urus::pubsub`, phoenix_pubsub-shaped, local node only (no distribution — -smarm has none): +smarm has none). All six design questions ratified by the user pre-code +("push" on the stated leans); as-landed: -- `subscribe(topic)`, `unsubscribe(topic)`, `broadcast(topic, msg)`, - `broadcast_from(self, topic, msg)`. -- Topic table: gen_server owning `HashMap>)>>`; - shard N ways by topic hash if the single server measures hot (don't - pre-shard; bench first — smarm channel sends are cheap). -- Dead subscriber cleanup via `monitor` + `handle_down` (this is exactly - what 06-10's gen_server `handle_down` was built for). -- `Arc` payloads: one allocation per broadcast, not per subscriber — - the shared-heap advantage over BEAM, take it. +- **Generic `PubSub`** per instance (Q1); heterogeneous buses are an + app-side enum. `Arc` payloads: one allocation per broadcast. +- **`subscribe(topic) -> Receiver>`** (Q2), fresh channel per + subscription; `subscribe_as(pid, ..)` pins lifetime to an explicit pid + for relay shapes. Sender-passing fan-in (`subscribe_with`) deliberately + deferred — compatible addition. +- **Unbounded mailboxes** (Q3): broadcast never blocks the table. Cleanup + is monitor `Down` (eager, on subscriber death) + prune-on-send-failure + (lazy, dropped receivers). Bench before sharding. +- **User-owned handle** (Q4). The ratified register-by-name helper turned + out NOT implementable: smarm's registry maps name → Pid and a Pid can't + be turned back into a `ServerRef`. Deferred pending smarm support (or a + type-erased global, rejected). `pid()` exposed. +- **Unique per `(pid, topic)`** (Q5): subscribe idempotent (replaces the + sender; the stale receiver closes). One monitor per live pid. + `handle_down` is a full table scan as ratified; pid→topics reverse + index is the documented optimization if deaths measure hot. +- Table: gen_server `HashMap>>>` — + single server, not sharded (bench first). `subscriber_count(topic)` + added (tests need to observe async cleanup; legitimate API). -Deliberately independent of HTTP — usable from any smarm app. Separate -module now, candidate for crate extraction later. +**Discoveries this cycle:** +- `PubSub::new()` is in-runtime only (spawns the table actor) but + pipelines are built pre-runtime → the composition pattern is a + NON-static `Arc>>` in the route closure. Non-static + is load-bearing: the drained pipeline drops the cell in-runtime, the + table's inbox closes, the table exits, AllDone is reachable. A `static` + (crud's store shape) hangs `serve_with_shutdown`. Corollary: relays + hold `Receiver` only, never a `PubSub` clone (mutual-keepalive cycle). + Proven by `shutdown_with_open_chat_terminates`. +- **`WsHandler::on_open(&mut self, sender)`** added (defaulted, + non-breaking): without it a listen-only ws client can never be + subscribed (`on_message` never fires). Same panic contract as + `on_message` (1011, no `on_close`). The 101 reaches the client before + `on_open` runs — subscription liveness needs an app-level ack + (documented; the integration tests use a `sync`/`synced` ack). +- hammer.sh default subset now includes `ws_`. + +Chunk 2's ws_chat: rooms as topics, `on_open` subscribes (conn-actor +pid) + spawns the relay, `broadcast_from(self_pid(), ..)` for no-echo. ## v0.6 — Channels diff --git a/examples/ws_chat.rs b/examples/ws_chat.rs new file mode 100644 index 0000000..841b773 --- /dev/null +++ b/examples/ws_chat.rs @@ -0,0 +1,128 @@ +//! WebSocket chat rooms (v0.5): `urus::pubsub` wired to the v0.4 duplex. +//! +//! cargo run --example ws_chat +//! websocat ws://127.0.0.1:8080/chat/lobby (run two of these) +//! +//! The shapes this demonstrates: +//! +//! - **One `PubSub` for the whole app**, topics are rooms +//! (`room:{name}`). Built lazily on the first connection via a +//! NON-static `Arc>` captured by the route closure: +//! `PubSub::new()` spawns the table actor, which smarm only allows +//! in-runtime — and keeping the cell non-static means the table's +//! last handle drops in-runtime when the drained pipeline drops, so +//! `serve_with_shutdown` actually returns. A `static` cell would pin +//! the table forever and block smarm's all-done. (Same constraint +//! crud's store solves with its OnceLock; the non-static refinement +//! is what makes graceful shutdown compose.) +//! +//! - **`on_open` subscribes and spawns the relay** — a listen-only +//! client receives the room without ever sending. The subscription is +//! pinned to the CONNECTION actor (`on_open` runs inside it), so the +//! monitor cleans up exactly when the connection dies. +//! +//! - **The relay holds the `Receiver` and a `WsSender` clone — and +//! deliberately NOT a `PubSub` handle.** A relay holding the handle +//! keeps the table's inbox open while the table keeps the relay's +//! receiver open: neither ever exits, and shutdown hangs. Receiver +//! only: conn dies → monitor prunes → sender drops → relay's recv +//! errs → relay exits. Every link in that chain is in-runtime. + +use std::sync::{Arc, OnceLock}; + +use urus::{ + serve_with_shutdown, shutdown_handle, Config, Conn, Message, Next, Pipeline, PubSub, Router, + WsHandler, WsSender, +}; + +type Bus = PubSub; + +struct ChatHandler { + bus: Arc>, + room: String, +} + +impl ChatHandler { + fn topic(&self) -> String { + format!("room:{}", self.room) + } + + fn bus(&self) -> &Bus { + // First connection anywhere spawns the table; we are inside the + // connection actor here, so the spawn is legal. + self.bus.get_or_init(PubSub::new) + } +} + +impl WsHandler for ChatHandler { + fn on_open(&mut self, sender: &WsSender) { + let topic = self.topic(); + let rx = match self.bus().subscribe(&topic) { + Ok(rx) => rx, + Err(_) => { + let _ = sender.close(1011, "chat bus down"); + return; + } + }; + let _ = self + .bus() + .broadcast_from(smarm::self_pid(), &topic, format!("* someone joined {topic}")); + + // The relay: room messages -> this socket. Exits when the + // subscription is pruned (conn death / unsubscribe) or the + // socket is gone (WsClosed). See the module docs for why it + // must not capture a Bus handle. + let out = sender.clone(); + smarm::spawn(move || { + while let Ok(msg) = rx.recv() { + if out.send(Message::Text((*msg).clone())).is_err() { + break; // client gone or server draining + } + } + }); + } + + fn on_message(&mut self, msg: Message, sender: &WsSender) { + let Message::Text(text) = msg else { + let _ = sender.close(1003, "text only"); + return; + }; + // broadcast_from: the sender's own relay is skipped — no echo. + // self_pid() here is the connection actor, the pid on_open + // subscribed as. + let _ = self + .bus() + .broadcast_from(smarm::self_pid(), &self.topic(), text); + } + + fn on_close(&mut self, _code: Option, _reason: &str) { + let topic = self.topic(); + let _ = self + .bus() + .broadcast_from(smarm::self_pid(), &topic, format!("* someone left {topic}")); + // No explicit unsubscribe: the connection actor is about to + // exit and the monitor prunes the subscription (which is also + // what stops the relay). + } +} + +fn main() { + let bus: Arc> = Arc::new(OnceLock::new()); + + let pipeline = Pipeline::new().plug(Router::new().get("/chat/:room", move |c: Conn, _n: Next| { + let room = c.params.get("room").unwrap_or("lobby").to_string(); + let bus = bus.clone(); + c.upgrade(ChatHandler { bus, room }) + })); + + let (handle, signal) = shutdown_handle(); + std::thread::spawn(move || { + println!("ws_chat on ws://127.0.0.1:8080/chat/:room — press Enter to shut down"); + let mut line = String::new(); + let _ = std::io::stdin().read_line(&mut line); + handle.shutdown(); + }); + + serve_with_shutdown(Config::new("0.0.0.0:8080".parse().unwrap()), pipeline, signal).unwrap(); + println!("ws_chat: drained, bye"); +} diff --git a/scripts/hammer.sh b/scripts/hammer.sh index 113541c..68685d4 100755 --- a/scripts/hammer.sh +++ b/scripts/hammer.sh @@ -18,7 +18,7 @@ export PATH="$HOME/.cargo/bin:$PATH" N=35 FULL=3 -SUBSET=(shutdown timeout reaped slowloris streaming chunked sse stalled) +SUBSET=(shutdown timeout reaped slowloris streaming chunked sse stalled ws_) while [[ $# -gt 0 ]]; do case "$1" in diff --git a/src/pubsub.rs b/src/pubsub.rs index 4991b5b..942ec12 100644 --- a/src/pubsub.rs +++ b/src/pubsub.rs @@ -38,6 +38,23 @@ //! but stays alive keeps its (one-shot, inert) monitor until death; //! that's a bounded bookkeeping entry, not a leak. //! +//! # Construction must happen in-runtime +//! +//! [`PubSub::new`] spawns the table actor, which smarm only permits from +//! inside `Runtime::run`. Pipelines are built *before* `serve*` boots the +//! runtime, so the working pattern (same constraint crud's store hits) is +//! lazy init from the first handler — but with a **non-static** +//! `Arc>>` captured by the route closure, NOT a +//! `static`: when the drained pipeline drops at graceful shutdown, the +//! cell (and thus the last handle) drops in-runtime, the table's inbox +//! closes, and the table exits — `serve_with_shutdown` returns. A +//! `static` pins the table forever and blocks smarm's all-done. The same +//! reasoning forbids long-lived consumer actors (relays, producers) from +//! holding a `PubSub` clone: relay holds table's inbox open, table holds +//! relay's receiver open, neither exits. Relays take the `Receiver` only. +//! See `examples/ws_chat.rs` and the `shutdown_with_open_chat_terminates` +//! integration test for the full chain. +//! //! # Why there is no `register(name)` helper (yet) //! //! smarm's registry maps `name → Pid`, but a `Pid` cannot be turned back diff --git a/src/ws/duplex.rs b/src/ws/duplex.rs index c1276dd..d1d368a 100644 --- a/src/ws/duplex.rs +++ b/src/ws/duplex.rs @@ -55,6 +55,22 @@ use std::time::Instant; /// error` close frame and skips `on_close` (the handler is not re-entered /// after it panicked). pub trait WsHandler: Send + 'static { + /// The connection is up: the 101 has been written and the frame loop + /// is about to start. Runs exactly once, first, inside the + /// connection actor. This is the place to subscribe / spawn + /// producers for sessions where the client may never send (a + /// listen-only chat member, a live feed) — `on_message` never fires + /// for those. NOTE the 101 reaches the client *before* `on_open` + /// runs: a client that needs to know its subscriptions are live must + /// use an application-level ack (send something, await the reply — + /// any reply proves `on_open` completed, since callbacks are + /// sequential in the conn actor). A panic here closes `1011` and + /// skips `on_close`, exactly like a panicking `on_message`. + /// Default: no-op. + fn on_open(&mut self, sender: &WsSender) { + let _ = sender; + } + /// A complete data message (fragments already reassembled, text /// already UTF-8 validated) arrived from the client. fn on_message(&mut self, msg: Message, sender: &WsSender); @@ -158,6 +174,16 @@ pub(crate) fn run_duplex( let sender = WsSender { tx }; let mut asm = Assembler::new(limits.max_message_bytes); + // on_open before anything is decoded — even a pipelined first frame + // is observed by a handler that knows the connection exists. Same + // panic contract as on_message: re-raise a stop sentinel, else 1011. + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handler.on_open(&sender))); + if r.is_err() { + smarm::preempt::check_cancelled(); + let _ = close_and_drain_code(raw, 1011, "", &mut buf, limits); + return; + } + loop { // ----- 1. Decode everything already buffered. ----- // Runs before the first select so a pipelined first frame is diff --git a/tests/integration.rs b/tests/integration.rs index 756d7ce..fa401a0 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1280,3 +1280,131 @@ fn shutdown_force_stops_open_ws() { let mut rest = Vec::new(); let _ = s.read_to_end(&mut rest); // EOF or reset; both fine } + +// --------------------------------------------------------------------------- +// v0.5: pubsub wired to ws (the chat shape) — see examples/ws_chat.rs +// --------------------------------------------------------------------------- + +/// Minimal chat handler mirroring the example: on_open subscribes (conn +/// actor pid) + spawns a relay holding ONLY the Receiver and a WsSender +/// clone; on_message broadcast_from's, skipping the sender's own relay. +struct Chat { + bus: std::sync::Arc>>, +} + +impl urus::WsHandler for Chat { + fn on_open(&mut self, sender: &urus::WsSender) { + let bus = self.bus.get_or_init(urus::PubSub::new); + let rx = bus.subscribe("room").unwrap(); + let _ = bus.broadcast_from(smarm::self_pid(), "room", "joined".into()); + let out = sender.clone(); + smarm::spawn(move || { + while let Ok(msg) = rx.recv() { + if out.send(urus::Message::Text((*msg).clone())).is_err() { + break; + } + } + }); + } + + fn on_message(&mut self, msg: urus::Message, sender: &urus::WsSender) { + if let urus::Message::Text(t) = msg { + // "sync" is the join-ack: a direct (non-broadcast) reply + // proving on_open — and therefore this client's + // subscription — completed. The tests need it because the + // 101 reaches the client BEFORE the conn actor runs + // on_open; without an ack, "subscribed yet?" is a race. + if t == "sync" { + let _ = sender.send(urus::Message::Text("synced".into())); + return; + } + let bus = self.bus.get_or_init(urus::PubSub::new); + let _ = bus.broadcast_from(smarm::self_pid(), "room", t); + } + } +} + +fn chat_pipeline() -> Pipeline { + let bus: std::sync::Arc>> = + std::sync::Arc::new(std::sync::OnceLock::new()); + Pipeline::new().plug(Router::new().get("/ws", move |c: Conn, _n: Next| { + let bus = bus.clone(); + c.upgrade(Chat { bus }) + })) +} + +/// Two clients in a room: a broadcast reaches the other client and (via +/// broadcast_from) never echoes to the sender. The on_open hook is what +/// makes the listen-only direction work at all — B receives without +/// having sent a single frame when A's first message arrives. +#[test] +fn ws_chat_broadcast_reaches_other_client_not_sender() { + let port = spawn_server(chat_pipeline()); + + let mut a = ws_connect(port); + let mut a_buf = Vec::new(); + // Join-ack A before B connects: A's sub must predate B's join + // notice for the read sequence below to be deterministic. + ws_send(&mut a, &Frame::new(Opcode::Text, "sync")); + assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"synced"); + + let mut b = ws_connect(port); + let mut b_buf = Vec::new(); + ws_send(&mut b, &Frame::new(Opcode::Text, "sync")); + assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"synced"); + + // A sees B's join notice — proves A's relay is live and B's + // subscription (made in on_open, before any client frame) is in. + assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"joined"); + + // A speaks; B receives it as its FIRST broadcast frame. + ws_send(&mut a, &Frame::new(Opcode::Text, "hello")); + assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"hello"); + + // The no-echo proof without timeout reads: B speaks, and A's next + // frame must be B's message — if A had been echoed its own "hello", + // that would arrive first. + ws_send(&mut b, &Frame::new(Opcode::Text, "yo")); + assert_eq!( + ws_read_frame(&mut a, &mut a_buf).payload, + b"yo", + "sender was echoed its own broadcast_from message" + ); +} + +/// THE structural test for the v0.5 shutdown chain: with two open chat +/// sockets, live relays, and a live pubsub table, graceful shutdown must +/// still terminate. Chain under test: drain force-stops conn actors → +/// monitors prune subscriptions (relay senders drop, relays' recv errs) +/// → listeners stop → last Arc drops the non-static bus cell +/// in-runtime → table inbox closes → table exits → AllDone → serve +/// returns. A static bus, or a relay holding a PubSub clone, hangs here. +#[test] +fn shutdown_with_open_chat_terminates() { + let (port, handle, done_rx) = + spawn_server_with_handle(chat_pipeline(), Duration::from_millis(300)); + + let mut a = ws_connect(port); + let mut a_buf = Vec::new(); + ws_send(&mut a, &Frame::new(Opcode::Text, "sync")); + assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"synced"); + + let mut b = ws_connect(port); + let mut b_buf = Vec::new(); + ws_send(&mut b, &Frame::new(Opcode::Text, "sync")); + assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"synced"); + + // Prove the full pubsub path is live before shutting down. + assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"joined"); + ws_send(&mut b, &Frame::new(Opcode::Text, "pre-shutdown")); + assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"pre-shutdown"); + + handle.shutdown(); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("serve did not return: pubsub table or a relay outlived the drain"); + + let mut rest = Vec::new(); + let _ = a.read_to_end(&mut rest); + let _ = b.read_to_end(&mut rest); +}