From 0de2baa72d4f5ee694edd8fe6e5eb501f88583a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 12 Jun 2026 15:22:54 +0000 Subject: [PATCH] =?UTF-8?q?feat(channels):=20opt-in=20session=20persistenc?= =?UTF-8?q?e=20=E2=80=94=20v0.6=20chunk=203?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChannelSession

(src/channels/session.rs): a channel actor that outlives its transport, buffering outbound broadcasts as the same Arc>s the relay path carries (zero re-allocation) until rejoin, buffer cap, or TTL. Registered per pattern via PrefixRouter::channel_session::(pattern, factory) / channel_session_default::(pattern); one registry gen_server per registration, monomorphic over S::Key (no type erasure). Machinery: deploy() computes the key on the conn actor and casts the join handshake to the registry; the registry maps key -> (pid, control sender), forwards reattaches, spawns-and-monitors fresh actors, and prunes on Down (pid-guarded against replaced entries). The session actor selects [inbound, bus, ctl] attached and select_timeout([ctl, bus], ttl-remaining) detached; detach triggers are the closed inbound arm and ws send failure (the failed broadcast is the buffer's first entry — nothing lost). Drain re-encodes under the new join generation. As-landed decisions (each in module docs): - ChannelFactory gained a #[doc(hidden)] deploy() seam (default = ephemeral linked actor, the chunk-1 behavior verbatim); JoinHandshake and ChannelInbox are opaque pub structs. Channel construction moved from run_channel into deploy — same linked blast radius. - Every attach calls ch.join() again (rejoin acks need a payload and channels get to re-auth); session channels must treat join as re-entrant. - A rejected rejoin ENDS the session (no zombie sessions held open for unauthorized clients); next join is a cold start. - A second transport EVICTS the first (best-effort phx_close); a session is single-transport. - Explicit leave ends the session, not just the attachment. - TTL per detach episode; subscribe once, on first successful join; cap.max(1) semantics (cap 0 = first buffered message tears down, but an empty buffer still waits out the TTL). - Key: Clone beyond the ratified Eq+Hash+Send — the registry keeps a pid -> key reverse index for Down pruning. - The registry spawns eagerly at registration (channel_session is in-runtime-only, same law as ChannelHub::new): a lazy OnceLock would block std-sync under green threads on first-join races. - Shutdown composes WITHOUT links: conns die -> router Arcs drop -> registry inbox closes -> registry exits -> ctl senders drop -> detached sessions wake on the closed ctl arm, terminate, exit. shutdown_with_detached_session_terminates is the proof (ttl 30s vs 5s budget — only the chain can reap it). Tests: 6 session unit tests (buffer-and-drain ordering across reconnect, TTL expiry, cap teardown, leave, eviction incl. stale conn-entry self-prune, rejected rejoin) + the shutdown integration test. Shared toy codec + conn harness extracted to channels/testkit.rs (mod.rs tests now import it; no behavior change). Hammer: 35x lifecycle subset (+channels +session) + 3x full (phoenix) + 1x full (phoenix+smarm-trace) + featureless + channels-only, all green; dbg-grep clean. Suite: 90 unit + 46 integration + 2 doc under --features phoenix. --- src/channels/mod.rs | 190 +++++------ src/channels/session.rs | 736 ++++++++++++++++++++++++++++++++++++++++ src/channels/testkit.rs | 121 +++++++ src/lib.rs | 2 +- tests/integration.rs | 57 +++- 5 files changed, 990 insertions(+), 116 deletions(-) create mode 100644 src/channels/session.rs create mode 100644 src/channels/testkit.rs diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 5973e04..b85e162 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -73,6 +73,12 @@ use std::sync::Arc; #[cfg(feature = "phoenix")] pub mod phoenix; +mod session; +pub use session::ChannelSession; + +#[cfg(test)] +pub(crate) mod testkit; + // --------------------------------------------------------------------------- // Wire-neutral frames // --------------------------------------------------------------------------- @@ -192,8 +198,43 @@ pub trait Channel: Send + 'static { /// possible (`"room:lobby"` arrives whole; pull `"lobby"` out yourself). pub trait ChannelFactory: Send + Sync { fn create(&self, topic: &str) -> Box>; + + /// How an accepted-routing join becomes a running channel actor. + /// Internal seam — the default (ephemeral actor, linked to the + /// connection, cold start per join) is the contract; only the + /// session machinery overrides it. Not part of the public API. + #[doc(hidden)] + fn deploy(&self, hs: JoinHandshake

) -> ChannelInbox

+ where + P: Encode + Decode, + { + let ch = self.create(&hs.topic); + let conn_pid = smarm::self_pid(); + let (tx, rx) = smarm::channel::>(); + smarm::spawn(move || { + run_channel(conn_pid, ch, hs.topic, hs.join_ref, hs.reference, hs.payload, hs.ws, hs.bus, rx) + }); + ChannelInbox(tx) + } } +/// Everything the conn-side handler knows about one `phx_join`, handed +/// to [`ChannelFactory::deploy`]. Opaque: fields are crate-internal. +#[doc(hidden)] +pub struct JoinHandshake { + topic: String, + join_ref: Option, + reference: Option, + payload: P, + ws: WsSender, + bus: PubSub>, +} + +/// The conn-side handler's sender into a deployed channel actor. +/// Opaque: the field is crate-internal. +#[doc(hidden)] +pub struct ChannelInbox(Sender>); + impl ChannelFactory

for F where F: Fn(&str) -> Box> + Send + Sync, @@ -257,6 +298,31 @@ impl PrefixRouter

{ { self.channel(pattern, DefaultFactory::(PhantomData)) } + + /// [`channel`](Self::channel), with opt-in session persistence + /// keyed and configured by `S`'s [`ChannelSession`] impl (usually + /// `S` is the channel type itself). **In-runtime only**: this + /// spawns the pattern's session-registry actor, same law as + /// [`ChannelHub::new`]. + pub fn channel_session(self, pattern: &str, factory: impl ChannelFactory

+ 'static) -> Self + where + S: ChannelSession

, + S::Key: Clone, + P: Encode + Decode, + { + self.channel(pattern, session::SessionFactory::new::(Arc::new(factory))) + } + + /// [`channel_session`](Self::channel_session) with a + /// [`Default`]-built impl that is its own session config. + pub fn channel_session_default(self, pattern: &str) -> Self + where + C: Channel

+ ChannelSession

+ Default, + C::Key: Clone, + P: Encode + Decode, + { + self.channel_session::(pattern, DefaultFactory::(PhantomData)) + } } impl TopicRouter

for PrefixRouter

{ @@ -495,18 +561,16 @@ impl WsHandler for SocketHandler

); return; }; - let (tx, rx) = smarm::channel::>(); - let conn_pid = smarm::self_pid(); - let topic = frame.topic.clone(); - let join_ref = frame.join_ref.clone(); - let reference = frame.reference; - let ws = sender.clone(); - let bus = self.bus.clone(); - smarm::spawn(move || { - run_channel(conn_pid, factory, topic, join_ref, reference, payload, ws, bus, rx) + let inbox = factory.deploy(JoinHandshake { + topic: frame.topic.clone(), + join_ref: frame.join_ref.clone(), + reference: frame.reference, + payload, + ws: sender.clone(), + bus: self.bus.clone(), }); self.joined - .insert(frame.topic, Joined { join_ref: frame.join_ref, tx }); + .insert(frame.topic, Joined { join_ref: frame.join_ref, tx: inbox.0 }); } EV_LEAVE => match self.joined.remove(&frame.topic) { Some(j) => { @@ -567,7 +631,7 @@ impl WsHandler for SocketHandler

#[allow(clippy::too_many_arguments)] fn run_channel( conn_pid: Pid, - factory: Arc>, + mut ch: Box>, topic: String, join_ref: Option, reference: Option, @@ -590,7 +654,6 @@ fn run_channel( join_ref, cur_ref: RefCell::new(None), }; - let mut ch = factory.create(&topic); socket.set_ref(reference); match ch.join(&topic, payload, &socket) { @@ -694,65 +757,11 @@ fn channel_loop( #[cfg(test)] mod tests { use super::*; - use crate::ws::duplex::Outbound; + use super::testkit::{eventually, Harness, TP}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; use std::time::Duration; - /// Toy payload + codec: `jref|ref|topic|kind|event-or-status|payload` - /// with `-` for absent refs. Enough to drive the machinery. - #[derive(Debug, Clone, PartialEq, Eq)] - struct TP(String); - - fn part(o: Option<&str>) -> &str { - o.unwrap_or("-") - } - - impl Encode for TP { - fn encode(f: FrameRef<'_, Self>) -> Message { - let s = match f.message { - MessageRef::Event { event, payload } => format!( - "{}|{}|{}|event|{}|{}", - part(f.join_ref), - part(f.reference), - f.topic, - event, - payload.map(|p| p.0.as_str()).unwrap_or("") - ), - MessageRef::Reply { status, payload } => format!( - "{}|{}|{}|reply|{}|{}", - part(f.join_ref), - part(f.reference), - f.topic, - match status { - Status::Ok => "ok", - Status::Error => "error", - }, - payload.map(|p| p.0.as_str()).unwrap_or("") - ), - }; - Message::Text(s) - } - } - - impl Decode for TP { - fn decode(msg: &Message) -> Result, DecodeError> { - let Message::Text(s) = msg else { - return Err(DecodeError("binary".into())); - }; - let p: Vec<&str> = s.splitn(6, '|').collect(); - if p.len() != 6 || p[3] != "event" { - return Err(DecodeError(format!("bad frame: {s}"))); - } - let opt = |x: &str| (x != "-").then(|| x.to_string()); - Ok(ChannelFrame { - join_ref: opt(p[0]), - reference: opt(p[1]), - topic: p[2].to_string(), - message: ChannelMessage::Event { event: p[4].to_string(), payload: TP(p[5].into()) }, - }) - } - } /// A room channel: rejects topic "room:locked"; echoes events back as /// replies; broadcasts on "shout"; flags terminate. @@ -789,44 +798,6 @@ mod tests { } } - /// A fake socket: the conn-side handler plus the raw Outbound channel - /// a real duplex loop would drain. `recv_text` pulls the next encoded - /// frame the "client" would see. - struct Harness { - handler: SocketHandler, - sender: WsSender, - out_rx: smarm::Receiver, - } - - impl Harness { - fn new(hub: &ChannelHub) -> Self { - let (tx, out_rx) = smarm::channel(); - Self { - handler: SocketHandler { - bus: hub.bus.clone(), - router: hub.router.clone(), - joined: HashMap::new(), - }, - sender: WsSender { tx }, - out_rx, - } - } - - /// Feed one wire line in, as the duplex loop would. - fn send(&mut self, line: &str) { - self.handler - .on_message(Message::Text(line.to_string()), &self.sender.clone()); - } - - /// Next encoded frame the client would see (2s budget — channel - /// actors answer asynchronously). - fn recv(&self) -> String { - match self.out_rx.recv_timeout(Duration::from_secs(2)) { - Ok(Outbound::Msg(Message::Text(s))) => s, - other => panic!("expected outbound text frame, got {:?}", other.is_ok()), - } - } - } fn hub(terminated: &Arc) -> ChannelHub { ChannelHub::new( @@ -834,15 +805,6 @@ mod tests { ) } - fn eventually(mut f: impl FnMut() -> bool) -> bool { - for _ in 0..200 { - if f() { - return true; - } - smarm::sleep(Duration::from_millis(10)); - } - false - } #[test] fn prefix_router_exact_and_prefix_and_miss() { diff --git a/src/channels/session.rs b/src/channels/session.rs new file mode 100644 index 0000000..20554c3 --- /dev/null +++ b/src/channels/session.rs @@ -0,0 +1,736 @@ +//! 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, ServerCtx}; +use smarm::{Down, ServerRef, 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: ServerRef>, +} + +/// 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 = (); + + fn init(&mut self, ctx: &ServerCtx) { + 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"); + } +} diff --git a/src/channels/testkit.rs b/src/channels/testkit.rs new file mode 100644 index 0000000..7cc2dc5 --- /dev/null +++ b/src/channels/testkit.rs @@ -0,0 +1,121 @@ +//! Shared test scaffolding for the channels module tree: the toy +//! pipe-delimited codec (`TP`) and the conn-side `Harness` that fakes a +//! websocket transport. `cfg(test)` only. + +use super::*; +use crate::ws::duplex::Outbound; +use std::time::Duration; + +/// Toy payload + codec: `jref|ref|topic|kind|event-or-status|payload` +/// with `-` for absent refs. Enough to drive the machinery. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct TP(pub(crate) String); + +pub(crate) fn part(o: Option<&str>) -> &str { + o.unwrap_or("-") +} + +impl Encode for TP { + fn encode(f: FrameRef<'_, Self>) -> Message { + let s = match f.message { + MessageRef::Event { event, payload } => format!( + "{}|{}|{}|event|{}|{}", + part(f.join_ref), + part(f.reference), + f.topic, + event, + payload.map(|p| p.0.as_str()).unwrap_or("") + ), + MessageRef::Reply { status, payload } => format!( + "{}|{}|{}|reply|{}|{}", + part(f.join_ref), + part(f.reference), + f.topic, + match status { + Status::Ok => "ok", + Status::Error => "error", + }, + payload.map(|p| p.0.as_str()).unwrap_or("") + ), + }; + Message::Text(s) + } +} + +impl Decode for TP { + fn decode(msg: &Message) -> Result, DecodeError> { + let Message::Text(s) = msg else { + return Err(DecodeError("binary".into())); + }; + let p: Vec<&str> = s.splitn(6, '|').collect(); + if p.len() != 6 || p[3] != "event" { + return Err(DecodeError(format!("bad frame: {s}"))); + } + let opt = |x: &str| (x != "-").then(|| x.to_string()); + Ok(ChannelFrame { + join_ref: opt(p[0]), + reference: opt(p[1]), + topic: p[2].to_string(), + message: ChannelMessage::Event { event: p[4].to_string(), payload: TP(p[5].into()) }, + }) + } +} + +/// A fake socket: the conn-side handler plus the raw Outbound channel +/// a real duplex loop would drain. `recv_text` pulls the next encoded +/// frame the "client" would see. +pub(crate) struct Harness { + handler: SocketHandler, + sender: WsSender, + out_rx: smarm::Receiver, +} + +impl Harness { + pub(crate) fn new(hub: &ChannelHub) -> Self { + let (tx, out_rx) = smarm::channel(); + Self { + handler: SocketHandler { + bus: hub.bus.clone(), + router: hub.router.clone(), + joined: HashMap::new(), + }, + sender: WsSender { tx }, + out_rx, + } + } + + /// Feed one wire line in, as the duplex loop would. + pub(crate) fn send(&mut self, line: &str) { + self.handler + .on_message(Message::Text(line.to_string()), &self.sender.clone()); + } + + /// Assert nothing arrives for this client within a short window — + /// for "the broadcast must NOT reach the evicted transport" cases. + pub(crate) fn assert_silence(&self) { + if let Ok(Outbound::Msg(Message::Text(s))) = + self.out_rx.recv_timeout(Duration::from_millis(100)) + { + panic!("expected silence, got frame: {s}"); + } + } + + /// Next encoded frame the client would see (2s budget — channel + /// actors answer asynchronously). + pub(crate) fn recv(&self) -> String { + match self.out_rx.recv_timeout(Duration::from_secs(2)) { + Ok(Outbound::Msg(Message::Text(s))) => s, + other => panic!("expected outbound text frame, got {:?}", other.is_ok()), + } + } +} + +pub(crate) fn eventually(mut f: impl FnMut() -> bool) -> bool { + for _ in 0..200 { + if f() { + return true; + } + smarm::sleep(Duration::from_millis(10)); + } + false +} diff --git a/src/lib.rs b/src/lib.rs index 2806104..e71d74c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,7 +39,7 @@ pub use router::Router; pub use sse::{EventSender, SseClosed}; pub use pubsub::{PubSub, PubSubDown}; #[cfg(feature = "channels")] -pub use channels::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status, TopicRouter}; +pub use channels::{Channel, ChannelHub, ChannelSession, ChannelSocket, PrefixRouter, Status, TopicRouter}; pub use ws::{Message, WsClosed, WsHandler, WsSender}; pub use serve::{ serve, serve_with, serve_with_shutdown, shutdown_handle, Config, Handle, ShutdownSignal, diff --git a/tests/integration.rs b/tests/integration.rs index d2636e9..42e2cf3 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1418,7 +1418,7 @@ mod channels_wire { use super::*; use serde_json::{json, Value}; use urus::channels::phoenix::Json; - use urus::channels::{Channel, ChannelHub, ChannelSocket, PrefixRouter, Status}; + use urus::channels::{Channel, ChannelHub, ChannelSession, ChannelSocket, PrefixRouter, Status}; type P = Json; @@ -1603,4 +1603,59 @@ mod channels_wire { let mut rest = Vec::new(); let _ = a.read_to_end(&mut rest); } + + /// Sessions: keyed by topic, default cap/ttl. The 30s ttl is the + /// test's lever — far past the assertion budget, so only the + /// registry-drop chain can reap a detached session at shutdown. + struct LobbySession; + impl ChannelSession

for LobbySession { + type Key = String; + fn session_key(topic: &str, _p: &P) -> String { + topic.to_owned() + } + } + + fn session_pipeline() -> Pipeline { + let hub: std::sync::Arc>> = + std::sync::Arc::new(std::sync::OnceLock::new()); + Pipeline::new().plug(Router::new().get("/socket", move |c: Conn, _n: Next| { + let hub = hub.clone(); + let hub = hub.get_or_init(|| { + ChannelHub::new(PrefixRouter::new().channel_session::( + "room:*", + |_: &str| Box::new(Lobby) as Box>, + )) + }); + hub.upgrade(c) + })) + } + + #[test] + fn shutdown_with_detached_session_terminates() { + // The session variant of the exit chain: the actor is NOT + // linked to its conn and survives the transport on purpose. + // Conn actors die at drain -> their router Arcs drop -> the + // registry's inbox closes -> registry exits -> control senders + // drop -> the detached session wakes on its closed control arm, + // terminates, and exits -> AllDone. No links anywhere in that + // chain; this test is the proof it composes. + let (port, handle, done_rx) = + spawn_server_with_handle(session_pipeline(), Duration::from_millis(300)); + + let mut a = chan_connect(port); + let mut a_buf = Vec::new(); + chan_send(&mut a, json!(["1", "1", "room:lobby", "phx_join", {}])); + let reply = chan_read(&mut a, &mut a_buf); + assert_eq!(reply[3], "phx_reply"); // joined: session actor live + + // Kill the transport and give the detach a moment to land, so + // shutdown meets a genuinely DETACHED actor (ttl 30s from now). + drop(a); + std::thread::sleep(Duration::from_millis(200)); + + handle.shutdown(); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("detached session outlived the drain: the registry-drop chain is broken"); + } }