//! Opt-in channel session persistence (the ratified v0.6 design). //! //! Default channel lifetime is transport-bound: cold start per join, //! actor dies with the socket. A [`ChannelSession`] impl registered via //! [`PrefixRouter::channel_session`](super::PrefixRouter::channel_session) //! changes that: the channel actor outlives its transport, buffering //! outbound broadcasts (`VecDeque>>` — the same `Arc`s //! the pubsub relay path carries, zero re-allocation) until the client //! rejoins, the buffer fills, or the TTL expires. //! //! Shape (one registry gen_server per `channel_session` registration, //! monomorphic over the session key — no type erasure): //! //! - `deploy` on the conn actor computes the session key and casts the //! whole join handshake to the registry. //! - The registry owns `key -> (pid, control sender)`. Existing entry: //! the handshake is forwarded as an `Attach` (reattach). Absent or //! dead (send failure raced the monitor `Down`): spawn fresh, //! monitor, insert. Down prunes, guarded by pid match against //! replaced entries. //! - The session actor is **not linked** to any connection — outliving //! the transport is the point. Its exits: explicit leave, rejected //! (re)join, TTL expiry, buffer cap, pubsub relay death, or the //! registry's control sender dropping — which is exactly the shutdown //! chain (conns die -> router `Arc`s drop -> registry's inbox closes //! -> registry exits -> control senders drop -> every parked session //! wakes on the closed control arm, terminates, and exits; `AllDone` //! composes without links). //! //! As-landed decisions (veto by diff): //! - **Every attach calls `ch.join()` again** on the same instance — //! the client's `phx_join` needs an ack payload and the channel gets //! to re-auth. Session channels must treat `join` as re-entrant. //! - **A rejected rejoin ends the session** (error reply, `terminate`, //! exit): `Err` means "this transport may not have this channel", and //! a zombie session held open for an unauthorized client is a //! liability. Next join is a cold start. //! - **A second transport evicts the first** (best-effort `phx_close` //! on the old socket): a session is single-transport by definition; //! two tabs wanting independent channels should not share a session //! key. //! - **TTL is per detach episode** (reset on every disconnect), and the //! pubsub subscription is made once, on the first successful join, to //! the first join's topic. //! - **`ChannelSession::Key: Clone`** beyond the ratified //! `Eq + Hash + Send` — the registry keeps a `pid -> key` reverse //! index for monitor-`Down` pruning. use super::*; use smarm::gen_server::{self, GenServer, GenServerCtx}; use smarm::{Down, GenServerRef, Watcher}; use std::collections::VecDeque; use std::hash::Hash; use std::time::{Duration, Instant}; /// Opt-in session persistence config for a channel type. All methods /// are static: the config is consulted before any channel instance /// exists. pub trait ChannelSession

: Send + 'static { /// What identifies "the same session" across reconnects. type Key: Eq + Hash + Send + 'static; /// Derive the session key from a join. Same key, same (live) /// channel actor; the join payload is the natural carrier for a /// client-held session token. fn session_key(topic: &str, join_payload: &P) -> Self::Key; /// Detached broadcasts buffered before the session tears itself /// down (next join is a cold start). fn buffer_cap() -> usize { 128 } /// How long a detached session waits for a rejoin before tearing /// itself down. Reset on every detach. fn ttl() -> Duration { Duration::from_secs(30) } } // --------------------------------------------------------------------------- // The session-aware factory (what channel_session registers) // --------------------------------------------------------------------------- pub(super) struct SessionFactory { inner: Arc>, keyfn: fn(&str, &P) -> K, registry: GenServerRef>, } /// The registry's working bounds for a session key. pub(super) trait SessionKey: Eq + Hash + Clone + Send + 'static {} impl SessionKey for K {} impl SessionFactory { /// **In-runtime only** — spawns the registry gen_server. pub(super) fn new>(inner: Arc>) -> Self { let registry = gen_server::start(Registry { factory: inner.clone(), cap: S::buffer_cap(), ttl: S::ttl(), sessions: HashMap::new(), pid_key: HashMap::new(), watcher: None, }); SessionFactory { inner, keyfn: S::session_key, registry } } } impl ChannelFactory

for SessionFactory { fn create(&self, topic: &str) -> Box> { self.inner.create(topic) } fn deploy(&self, hs: JoinHandshake

) -> ChannelInbox

where P: Encode + Decode, { let key = (self.keyfn)(&hs.topic, &hs.payload); let (tx, rx) = smarm::channel::>(); // Cast: the actor acks the join straight to the socket; the // conn actor has nothing to wait for. A dead registry can only // mean shutdown — the failed join is moot. let _ = self.registry.cast(Join { key, hs, inbound: rx }); ChannelInbox(tx) } } // --------------------------------------------------------------------------- // The registry gen_server: key -> live session actor // --------------------------------------------------------------------------- pub(super) struct Join { key: K, hs: JoinHandshake

, inbound: Receiver>, } struct Registry { factory: Arc>, cap: usize, ttl: Duration, sessions: HashMap>)>, /// Reverse index for `handle_down` — the reason `Key: Clone`. pid_key: HashMap, watcher: Option>>, } impl GenServer for Registry { type Call = (); type Reply = (); type Cast = Join; type Info = (); type Timer = (); fn init(&mut self, ctx: &GenServerCtx) { self.watcher = Some(ctx.watcher()); } fn handle_call(&mut self, _request: ()) {} fn handle_cast(&mut self, Join { key, hs, inbound }: Join) { let attach = Ctl::Attach { hs, inbound }; let attach = match self.sessions.get(&key) { Some((_, ctl)) => match ctl.send(attach) { Ok(()) => return, // reattached // The actor died and its Down hasn't landed yet: // recover the handshake, replace below. Err(smarm::channel::SendError(a)) => a, }, None => attach, }; let (ctl_tx, ctl_rx) = smarm::channel::>(); let (factory, cap, ttl) = (self.factory.clone(), self.cap, self.ttl); let pid = smarm::spawn(move || run_session(factory, cap, ttl, ctl_rx)).pid(); // Cannot fail: the receiver is owned by the just-spawned actor. let _ = ctl_tx.send(attach); self.watcher .as_ref() .expect("watcher set in init") .watch(smarm::monitor(pid)); self.pid_key.insert(pid, key.clone()); self.sessions.insert(key, (pid, ctl_tx)); } fn handle_down(&mut self, down: Down) { if let Some(key) = self.pid_key.remove(&down.pid) { // Pid guard: a stale Down for an entry handle_cast already // replaced must not evict the replacement. if self.sessions.get(&key).is_some_and(|(p, _)| *p == down.pid) { self.sessions.remove(&key); } } } } // --------------------------------------------------------------------------- // The session actor // --------------------------------------------------------------------------- enum Ctl { Attach { hs: JoinHandshake

, inbound: Receiver> }, } enum Step { Attach { hs: JoinHandshake

, inbound: Receiver> }, Attached { inbound: Receiver> }, Detached, Exit, } fn run_session( factory: Arc>, cap: usize, ttl: Duration, ctl: Receiver>, ) { // The initial Attach is cast by the registry in the same handler // that spawned us. Closed instead = registry died first (shutdown // raced the spawn): nothing exists yet, nothing to clean. let Ok(Ctl::Attach { hs, inbound }) = ctl.recv() else { return }; // Partial moves: ws/bus/topic/join_ref go into the socket, // reference/payload stay behind for the first attach below. let mut socket = ChannelSocket { ws: hs.ws, bus: hs.bus, topic: hs.topic, join_ref: hs.join_ref, cur_ref: RefCell::new(None), }; let mut ch: Option>> = None; let mut joined = false; let mut bus_rx: Option>>> = None; let mut buffer: VecDeque>> = VecDeque::new(); let mut step = attach( &factory, &mut ch, &mut joined, &mut bus_rx, &mut buffer, &mut socket, hs.reference, hs.payload, inbound, ); loop { step = match step { Step::Attach { hs, inbound } => { socket.ws = hs.ws; socket.join_ref = hs.join_ref; socket.topic = hs.topic; attach( &factory, &mut ch, &mut joined, &mut bus_rx, &mut buffer, &mut socket, hs.reference, hs.payload, inbound, ) } Step::Attached { inbound } => attached_loop( ch.as_deref_mut().expect("attached implies created"), &socket, &ctl, &inbound, bus_rx.as_ref().expect("attached implies subscribed"), &mut buffer, ), Step::Detached => detached_loop( &ctl, bus_rx.as_ref().expect("detached only after first subscribe"), &mut buffer, cap, ttl, ), Step::Exit => { if joined { if let Some(c) = ch.as_deref_mut() { c.terminate(); } } return; } }; } } /// One attach handshake: create-once, (re)join, subscribe-once, ack, /// drain. On a drain failure the undelivered tail stays in `buffer` /// for the next attach. #[allow(clippy::too_many_arguments)] fn attach( factory: &Arc>, ch: &mut Option>>, joined: &mut bool, bus_rx: &mut Option>>>, buffer: &mut VecDeque>>, socket: &mut ChannelSocket

, reference: Option, payload: P, inbound: Receiver>, ) -> Step

{ if ch.is_none() { *ch = Some(factory.create(&socket.topic)); } let c = ch.as_deref_mut().expect("just created"); socket.set_ref(reference); let topic = socket.topic.clone(); match c.join(&topic, payload, socket) { Ok(reply) => { if bus_rx.is_none() { // First successful join: subscribe BEFORE the ok reply // (subscribe is a call — once the client sees its ack, // membership is a fact, not a race). Once per session: // the subscription is actor-scoped, survives detach, // and that survival IS the buffering path. match socket.bus.subscribe(&topic) { Ok(rx) => *bus_rx = Some(rx), Err(_) => { socket.send_reply(Status::Error, None); socket.set_ref(None); return Step::Exit; } } } *joined = true; socket.send_reply(Status::Ok, Some(&reply)); socket.set_ref(None); // Drain with the NEW join generation's ref; buffered pushes // have no request to correlate to (reference: None). while let Some(b) = buffer.front() { let out = P::encode(FrameRef { join_ref: socket.join_ref.as_deref(), reference: None, topic: &socket.topic, message: MessageRef::Event { event: &b.event, payload: Some(&b.payload) }, }); if socket.ws.send(out).is_ok() { buffer.pop_front(); } else { return Step::Detached; } } Step::Attached { inbound } } Err(reply) => { // Rejected (re)join = session over, by decision: error // reply, terminate (in run_session, iff it ever joined), // exit. Next join is a cold start. socket.send_reply(Status::Error, Some(&reply)); socket.set_ref(None); Step::Exit } } } fn attached_loop( ch: &mut dyn Channel

, socket: &ChannelSocket

, ctl: &Receiver>, inbound: &Receiver>, bus_rx: &Receiver>>, buffer: &mut VecDeque>>, ) -> Step

{ // ctl sits at lowest priority: it only ever carries an eviction by // a second transport (rare) or the closed-arm shutdown signal. A // closed arm stays ready forever, but every closed observation // exits the loop immediately — no spin, no starvation window. loop { match smarm::select(&[inbound, bus_rx, ctl]) { 0 => match inbound.try_recv() { Ok(Some(In::Event { reference, event, payload })) => { socket.set_ref(reference); ch.handle_in(&event, payload, socket); socket.set_ref(None); } Ok(Some(In::Leave { reference })) => { // Explicit leave ends the SESSION, not just the // attachment: ok reply, close event, terminate. socket.set_ref(reference); socket.send_reply(Status::Ok, None); socket.set_ref(None); close_event(socket); return Step::Exit; } Ok(None) => {} // Conn actor finished or replaced this join generation: // the transport is gone, the session is not. Err(_) => return Step::Detached, }, 1 => match bus_rx.try_recv() { Ok(Some(b)) => { let out = P::encode(FrameRef { join_ref: socket.join_ref.as_deref(), reference: None, topic: &socket.topic, message: MessageRef::Event { event: &b.event, payload: Some(&b.payload) }, }); if socket.ws.send(out).is_err() { // Transport died under us: this broadcast is // the buffer's first entry, nothing was lost. buffer.push_back(b); return Step::Detached; } } Ok(None) => {} // The topic relay died: pubsub is gone, session over. Err(_) => return Step::Exit, }, 2 => match ctl.try_recv() { Ok(Some(Ctl::Attach { hs, inbound })) => { // A second transport claims the session key: evict // this one (best effort — it may already be dead) // and hand the session over. close_event(socket); return Step::Attach { hs, inbound }; } Ok(None) => {} // Registry gone = shutdown chain: terminate and exit. Err(_) => return Step::Exit, }, _ => unreachable!("three arms"), } } } fn detached_loop( ctl: &Receiver>, bus_rx: &Receiver>>, buffer: &mut VecDeque>>, cap: usize, ttl: Duration, ) -> Step

{ // TTL is per detach episode: the clock starts now, every time. let deadline = Instant::now() + ttl; loop { // cap.max(1): a cap of 0 means "no buffering tolerated" — the // first buffered broadcast tears the session down; an empty // buffer still waits out the TTL. if buffer.len() >= cap.max(1) { return Step::Exit; } let remaining = deadline.saturating_duration_since(Instant::now()); match smarm::select_timeout(&[ctl, bus_rx], remaining) { // TTL expired with no rejoin: session over. None => return Step::Exit, Some(0) => match ctl.try_recv() { Ok(Some(Ctl::Attach { hs, inbound })) => return Step::Attach { hs, inbound }, Ok(None) => {} Err(_) => return Step::Exit, }, Some(1) => match bus_rx.try_recv() { Ok(Some(b)) => buffer.push_back(b), Ok(None) => {} Err(_) => return Step::Exit, }, Some(_) => unreachable!("two arms"), } } } /// Best-effort `phx_close` push for the socket's current join /// generation. fn close_event(socket: &ChannelSocket

) { let _ = socket.ws.send(P::encode(FrameRef { join_ref: socket.join_ref.as_deref(), reference: None, topic: &socket.topic, message: MessageRef::Event { event: EV_CLOSE, payload: None }, })); } // --------------------------------------------------------------------------- // Tests — runtime-backed, on the toy pipe codec from testkit. The // session-config types (`Cfg*`) are deliberately separate from the // channel they configure: `channel_session::` only needs `S` for // keying/cap/ttl, the factory builds the channel. // --------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::super::testkit::{eventually, Harness, TP}; use super::super::*; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; use std::time::Duration; /// Counts its joins (cold start replies `joined:1`); flags /// terminate; broadcasts on "shout". struct Counter { joins: u32, term: Arc, reject_rejoin: bool, } impl Channel for Counter { fn join(&mut self, _topic: &str, _payload: TP, _s: &ChannelSocket) -> Result { self.joins += 1; if self.reject_rejoin && self.joins > 1 { return Err(TP("not again".into())); } Ok(TP(format!("joined:{}", self.joins))) } fn handle_in(&mut self, event: &str, payload: TP, s: &ChannelSocket) { match event { "shout" => s.broadcast("news", payload), _ => s.reply(Status::Ok, TP(format!("echo:{}", payload.0))), } } fn terminate(&mut self) { self.term.store(true, Ordering::SeqCst); } } fn counter_factory(term: &Arc, reject_rejoin: bool) -> impl ChannelFactory { let term = term.clone(); move |_t: &str| -> Box> { Box::new(Counter { joins: 0, term: term.clone(), reject_rejoin }) } } /// Session config: keyed by topic alone, defaults otherwise. struct Cfg; impl ChannelSession for Cfg { type Key = String; fn session_key(topic: &str, _p: &TP) -> String { topic.to_owned() } } /// Tiny TTL (50ms) for the expiry test. struct CfgTtl; impl ChannelSession for CfgTtl { type Key = String; fn session_key(topic: &str, _p: &TP) -> String { topic.to_owned() } fn ttl() -> Duration { Duration::from_millis(50) } } /// Cap of 2 (and a long TTL so only the cap can fire). struct CfgCap; impl ChannelSession for CfgCap { type Key = String; fn session_key(topic: &str, _p: &TP) -> String { topic.to_owned() } fn buffer_cap() -> usize { 2 } fn ttl() -> Duration { Duration::from_secs(30) } } fn session_hub>( term: &Arc, reject_rejoin: bool, ) -> ChannelHub { ChannelHub::new( PrefixRouter::new() .channel_session::("room:*", counter_factory(term, reject_rejoin)), ) } #[test] fn session_buffers_across_reconnect_and_drains_in_order() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); smarm::run(move || { let hub = session_hub::(&term, false); let mut h1 = Harness::new(&hub); h1.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h1.recv()); // Transport dies. Whether the actor has observed the detach // yet or not, both paths (closed inbound arm, failed ws // send) land the broadcasts in the buffer. drop(h1); hub.broadcast("room:a", "news", TP("one".into())); hub.broadcast("room:a", "news", TP("two".into())); let mut h2 = Harness::new(&hub); h2.send("j2|r2|room:a|event|phx_join|u"); out2.lock().unwrap().push(h2.recv()); // warm rejoin ack out2.lock().unwrap().push(h2.recv()); // drained, in order out2.lock().unwrap().push(h2.recv()); }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j2|r2|room:a|reply|ok|joined:2"); assert_eq!(out[2], "j2|-|room:a|event|news|one"); assert_eq!(out[3], "j2|-|room:a|event|news|two"); } #[test] fn session_ttl_expiry_tears_down_then_cold_start() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); let term2 = term.clone(); smarm::run(move || { let hub = session_hub::(&term2, false); let mut h1 = Harness::new(&hub); h1.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h1.recv()); drop(h1); assert!(eventually(|| term2.load(Ordering::SeqCst)), "ttl never fired"); let mut h2 = Harness::new(&hub); h2.send("j2|r2|room:a|event|phx_join|u"); out2.lock().unwrap().push(h2.recv()); }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j2|r2|room:a|reply|ok|joined:1"); // cold } #[test] fn session_buffer_cap_tears_down_then_cold_start() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); smarm::run(move || { let hub = session_hub::(&term, false); let mut h1 = Harness::new(&hub); h1.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h1.recv()); drop(h1); // Cap is 2: the second buffered broadcast tears it down // (ttl is 30s — only the cap can fire here). hub.broadcast("room:a", "news", TP("one".into())); hub.broadcast("room:a", "news", TP("two".into())); assert!(eventually(|| term.load(Ordering::SeqCst)), "cap never fired"); let mut h2 = Harness::new(&hub); h2.send("j2|r2|room:a|event|phx_join|u"); out2.lock().unwrap().push(h2.recv()); }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j2|r2|room:a|reply|ok|joined:1"); // cold } #[test] fn leave_ends_the_session_not_just_the_attachment() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); smarm::run(move || { let hub = session_hub::(&term, false); let mut h = Harness::new(&hub); h.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h.recv()); h.send("j1|r2|room:a|event|phx_leave|"); out2.lock().unwrap().push(h.recv()); // ok reply out2.lock().unwrap().push(h.recv()); // phx_close assert!(eventually(|| term.load(Ordering::SeqCst)), "leave never terminated"); h.send("j2|r3|room:a|event|phx_join|u"); out2.lock().unwrap().push(h.recv()); }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j1|r2|room:a|reply|ok|"); assert_eq!(out[2], "j1|-|room:a|event|phx_close|"); assert_eq!(out[3], "j2|r3|room:a|reply|ok|joined:1"); // cold } #[test] fn second_transport_evicts_first() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); smarm::run(move || { let hub = session_hub::(&term, false); let mut h1 = Harness::new(&hub); h1.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h1.recv()); let mut h2 = Harness::new(&hub); h2.send("j2|r2|room:a|event|phx_join|u"); out2.lock().unwrap().push(h2.recv()); // warm handover ack out2.lock().unwrap().push(h1.recv()); // eviction phx_close hub.broadcast("room:a", "news", TP("x".into())); out2.lock().unwrap().push(h2.recv()); // only h2 is attached h1.assert_silence(); // h1's conn-side entry is stale: its inbound receiver was // dropped on eviction, so the next event self-prunes with // an error reply. h1.send("j1|r3|room:a|event|ping|x"); out2.lock().unwrap().push(h1.recv()); }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j2|r2|room:a|reply|ok|joined:2"); assert_eq!(out[2], "j1|-|room:a|event|phx_close|"); assert_eq!(out[3], "j2|-|room:a|event|news|x"); assert_eq!(out[4], "j1|r3|room:a|reply|error|"); } #[test] fn rejected_rejoin_ends_the_session() { let out = Arc::new(Mutex::new(Vec::::new())); let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false))); smarm::run(move || { let hub = session_hub::(&term, true); let mut h1 = Harness::new(&hub); h1.send("j1|r1|room:a|event|phx_join|u"); out2.lock().unwrap().push(h1.recv()); drop(h1); let mut h2 = Harness::new(&hub); h2.send("j2|r2|room:a|event|phx_join|u"); out2.lock().unwrap().push(h2.recv()); // rejoin rejected assert!(eventually(|| term.load(Ordering::SeqCst)), "reject never terminated"); let mut h3 = Harness::new(&hub); h3.send("j3|r3|room:a|event|phx_join|u"); out2.lock().unwrap().push(h3.recv()); // cold start }); let out = out.lock().unwrap(); assert_eq!(out[0], "j1|r1|room:a|reply|ok|joined:1"); assert_eq!(out[1], "j2|r2|room:a|reply|error|not again"); assert_eq!(out[2], "j3|r3|room:a|reply|ok|joined:1"); } }