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

feat(channels): opt-in session persistence — v0.6 chunk 3

Browse Source 
ChannelSession<P> (src/channels/session.rs): a channel actor that
outlives its transport, buffering outbound broadcasts as the same
Arc<Broadcast<P>>s the relay path carries (zero re-allocation) until
rejoin, buffer cap, or TTL. Registered per pattern via
PrefixRouter::channel_session::<S>(pattern, factory) /
channel_session_default::<C>(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.
This commit is contained in:
Claude
2026-06-12 15:22:54 +00:00
parent 39e4f70e9d
commit 0de2baa72d
5 changed files with 990 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/channels/mod.rs
+76 -114
View File
@@ -73,6 +73,12 @@ use std::sync::Arc;
#[cfg(feature = "phoenix")] #[cfg(feature = "phoenix")]
pub mod phoenix; pub mod phoenix;
mod session;
pub use session::ChannelSession;
#[cfg(test)]
pub(crate) mod testkit;
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Wire-neutral frames // Wire-neutral frames
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -192,7 +198,42 @@ pub trait Channel<P: Send + Sync + 'static>: Send + 'static {
/// possible (`"room:lobby"` arrives whole; pull `"lobby"` out yourself). /// possible (`"room:lobby"` arrives whole; pull `"lobby"` out yourself).
pub trait ChannelFactory<P: Send + Sync + 'static>: Send + Sync { pub trait ChannelFactory<P: Send + Sync + 'static>: Send + Sync {
fn create(&self, topic: &str) -> Box<dyn Channel<P>>; fn create(&self, topic: &str) -> Box<dyn Channel<P>>;
/// 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<P>) -> ChannelInbox<P>
where
P: Encode + Decode,
{
let ch = self.create(&hs.topic);
let conn_pid = smarm::self_pid();
let (tx, rx) = smarm::channel::<In<P>>();
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<P: Send + Sync + 'static> {
topic: String,
join_ref: Option<String>,
reference: Option<String>,
payload: P,
ws: WsSender,
bus: PubSub<Broadcast<P>>,
}
/// The conn-side handler's sender into a deployed channel actor.
/// Opaque: the field is crate-internal.
#[doc(hidden)]
pub struct ChannelInbox<P: Send + Sync + 'static>(Sender<In<P>>);
impl<P: Send + Sync + 'static, F> ChannelFactory<P> for F impl<P: Send + Sync + 'static, F> ChannelFactory<P> for F
where where
@@ -257,6 +298,31 @@ impl<P: Send + Sync + 'static> PrefixRouter<P> {
{ {
self.channel(pattern, DefaultFactory::<C>(PhantomData)) self.channel(pattern, DefaultFactory::<C>(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<S>(self, pattern: &str, factory: impl ChannelFactory<P> + 'static) -> Self
where
S: ChannelSession<P>,
S::Key: Clone,
P: Encode + Decode,
{
self.channel(pattern, session::SessionFactory::new::<S>(Arc::new(factory)))
}
/// [`channel_session`](Self::channel_session) with a
/// [`Default`]-built impl that is its own session config.
pub fn channel_session_default<C>(self, pattern: &str) -> Self
where
C: Channel<P> + ChannelSession<P> + Default,
C::Key: Clone,
P: Encode + Decode,
{
self.channel_session::<C>(pattern, DefaultFactory::<C>(PhantomData))
}
} }
impl<P: Send + Sync + 'static> TopicRouter<P> for PrefixRouter<P> { impl<P: Send + Sync + 'static> TopicRouter<P> for PrefixRouter<P> {
@@ -495,18 +561,16 @@ impl<P: Encode + Decode + Send + Sync + 'static> WsHandler for SocketHandler<P>
); );
return; return;
}; };
let (tx, rx) = smarm::channel::<In<P>>(); let inbox = factory.deploy(JoinHandshake {
let conn_pid = smarm::self_pid(); topic: frame.topic.clone(),
let topic = frame.topic.clone(); join_ref: frame.join_ref.clone(),
let join_ref = frame.join_ref.clone(); reference: frame.reference,
let reference = frame.reference; payload,
let ws = sender.clone(); ws: sender.clone(),
let bus = self.bus.clone(); bus: self.bus.clone(),
smarm::spawn(move || {
run_channel(conn_pid, factory, topic, join_ref, reference, payload, ws, bus, rx)
}); });
self.joined 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) { EV_LEAVE => match self.joined.remove(&frame.topic) {
Some(j) => { Some(j) => {
@@ -567,7 +631,7 @@ impl<P: Encode + Decode + Send + Sync + 'static> WsHandler for SocketHandler<P>
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
fn run_channel<P: Encode + Decode + Send + Sync + 'static>( fn run_channel<P: Encode + Decode + Send + Sync + 'static>(
conn_pid: Pid, conn_pid: Pid,
factory: Arc<dyn ChannelFactory<P>>, mut ch: Box<dyn Channel<P>>,
topic: String, topic: String,
join_ref: Option<String>, join_ref: Option<String>,
reference: Option<String>, reference: Option<String>,
@@ -590,7 +654,6 @@ fn run_channel<P: Encode + Decode + Send + Sync + 'static>(
join_ref, join_ref,
cur_ref: RefCell::new(None), cur_ref: RefCell::new(None),
}; };
let mut ch = factory.create(&topic);
socket.set_ref(reference); socket.set_ref(reference);
match ch.join(&topic, payload, &socket) { match ch.join(&topic, payload, &socket) {
@@ -694,65 +757,11 @@ fn channel_loop<P: Encode + Decode + Send + Sync + 'static>(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::ws::duplex::Outbound; use super::testkit::{eventually, Harness, TP};
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex; use std::sync::Mutex;
use std::time::Duration; 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<ChannelFrame<Self>, 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 /// A room channel: rejects topic "room:locked"; echoes events back as
/// replies; broadcasts on "shout"; flags terminate. /// 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<TP>,
sender: WsSender,
out_rx: smarm::Receiver<Outbound>,
}
impl Harness {
fn new(hub: &ChannelHub<TP>) -> 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<AtomicBool>) -> ChannelHub<TP> { fn hub(terminated: &Arc<AtomicBool>) -> ChannelHub<TP> {
ChannelHub::new( 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] #[test]
fn prefix_router_exact_and_prefix_and_miss() { fn prefix_router_exact_and_prefix_and_miss() {
src/channels/session.rs
+736
View File
@@ -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<Arc<Broadcast<P>>>` — 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<P>: 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<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> {
inner: Arc<dyn ChannelFactory<P>>,
keyfn: fn(&str, &P) -> K,
registry: ServerRef<Registry<P, K>>,
}
/// The registry's working bounds for a session key.
pub(super) trait SessionKey: Eq + Hash + Clone + Send + 'static {}
impl<K: Eq + Hash + Clone + Send + 'static> SessionKey for K {}
impl<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> SessionFactory<P, K> {
/// **In-runtime only** — spawns the registry gen_server.
pub(super) fn new<S: ChannelSession<P, Key = K>>(inner: Arc<dyn ChannelFactory<P>>) -> 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<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> ChannelFactory<P>
for SessionFactory<P, K>
{
fn create(&self, topic: &str) -> Box<dyn Channel<P>> {
self.inner.create(topic)
}
fn deploy(&self, hs: JoinHandshake<P>) -> ChannelInbox<P>
where
P: Encode + Decode,
{
let key = (self.keyfn)(&hs.topic, &hs.payload);
let (tx, rx) = smarm::channel::<In<P>>();
// 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<P: Send + Sync + 'static, K> {
key: K,
hs: JoinHandshake<P>,
inbound: Receiver<In<P>>,
}
struct Registry<P: Send + Sync + 'static, K> {
factory: Arc<dyn ChannelFactory<P>>,
cap: usize,
ttl: Duration,
sessions: HashMap<K, (Pid, Sender<Ctl<P>>)>,
/// Reverse index for `handle_down` — the reason `Key: Clone`.
pid_key: HashMap<Pid, K>,
watcher: Option<Watcher>,
}
impl<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> GenServer for Registry<P, K> {
type Call = ();
type Reply = ();
type Cast = Join<P, K>;
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<P, K>) {
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::<Ctl<P>>();
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<P: Send + Sync + 'static> {
Attach { hs: JoinHandshake<P>, inbound: Receiver<In<P>> },
}
enum Step<P: Send + Sync + 'static> {
Attach { hs: JoinHandshake<P>, inbound: Receiver<In<P>> },
Attached { inbound: Receiver<In<P>> },
Detached,
Exit,
}
fn run_session<P: Encode + Decode + Send + Sync + 'static>(
factory: Arc<dyn ChannelFactory<P>>,
cap: usize,
ttl: Duration,
ctl: Receiver<Ctl<P>>,
) {
// 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<Box<dyn Channel<P>>> = None;
let mut joined = false;
let mut bus_rx: Option<Receiver<Arc<Broadcast<P>>>> = None;
let mut buffer: VecDeque<Arc<Broadcast<P>>> = 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<P: Encode + Decode + Send + Sync + 'static>(
factory: &Arc<dyn ChannelFactory<P>>,
ch: &mut Option<Box<dyn Channel<P>>>,
joined: &mut bool,
bus_rx: &mut Option<Receiver<Arc<Broadcast<P>>>>,
buffer: &mut VecDeque<Arc<Broadcast<P>>>,
socket: &mut ChannelSocket<P>,
reference: Option<String>,
payload: P,
inbound: Receiver<In<P>>,
) -> Step<P> {
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<P: Encode + Decode + Send + Sync + 'static>(
ch: &mut dyn Channel<P>,
socket: &ChannelSocket<P>,
ctl: &Receiver<Ctl<P>>,
inbound: &Receiver<In<P>>,
bus_rx: &Receiver<Arc<Broadcast<P>>>,
buffer: &mut VecDeque<Arc<Broadcast<P>>>,
) -> Step<P> {
// 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<P: Encode + Decode + Send + Sync + 'static>(
ctl: &Receiver<Ctl<P>>,
bus_rx: &Receiver<Arc<Broadcast<P>>>,
buffer: &mut VecDeque<Arc<Broadcast<P>>>,
cap: usize,
ttl: Duration,
) -> Step<P> {
// 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<P: Encode + Decode + Send + Sync + 'static>(socket: &ChannelSocket<P>) {
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::<S>` 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<AtomicBool>,
reject_rejoin: bool,
}
impl Channel<TP> for Counter {
fn join(&mut self, _topic: &str, _payload: TP, _s: &ChannelSocket<TP>) -> Result<TP, TP> {
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<TP>) {
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<AtomicBool>, reject_rejoin: bool) -> impl ChannelFactory<TP> {
let term = term.clone();
move |_t: &str| -> Box<dyn Channel<TP>> {
Box::new(Counter { joins: 0, term: term.clone(), reject_rejoin })
}
}
/// Session config: keyed by topic alone, defaults otherwise.
struct Cfg;
impl ChannelSession<TP> 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<TP> 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<TP> 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<S: ChannelSession<TP, Key = String>>(
term: &Arc<AtomicBool>,
reject_rejoin: bool,
) -> ChannelHub<TP> {
ChannelHub::new(
PrefixRouter::new()
.channel_session::<S>("room:*", counter_factory(term, reject_rejoin)),
)
}
#[test]
fn session_buffers_across_reconnect_and_drains_in_order() {
let out = Arc::new(Mutex::new(Vec::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
smarm::run(move || {
let hub = session_hub::<Cfg>(&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::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
let term2 = term.clone();
smarm::run(move || {
let hub = session_hub::<CfgTtl>(&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::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
smarm::run(move || {
let hub = session_hub::<CfgCap>(&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::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
smarm::run(move || {
let hub = session_hub::<Cfg>(&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::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
smarm::run(move || {
let hub = session_hub::<Cfg>(&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::<String>::new()));
let (out2, term) = (out.clone(), Arc::new(AtomicBool::new(false)));
smarm::run(move || {
let hub = session_hub::<Cfg>(&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");
}
}
src/channels/testkit.rs
+121
View File
@@ -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<ChannelFrame<Self>, 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<TP>,
sender: WsSender,
out_rx: smarm::Receiver<Outbound>,
}
impl Harness {
pub(crate) fn new(hub: &ChannelHub<TP>) -> 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
}
src/lib.rs
+1 -1
View File
@@ -39,7 +39,7 @@ pub use router::Router;
pub use sse::{EventSender, SseClosed}; pub use sse::{EventSender, SseClosed};
pub use pubsub::{PubSub, PubSubDown}; pub use pubsub::{PubSub, PubSubDown};
#[cfg(feature = "channels")] #[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 ws::{Message, WsClosed, WsHandler, WsSender};
pub use serve::{ pub use serve::{
serve, serve_with, serve_with_shutdown, shutdown_handle, Config, Handle, ShutdownSignal, serve, serve_with, serve_with_shutdown, shutdown_handle, Config, Handle, ShutdownSignal,
tests/integration.rs
+56 -1
View File
@@ -1418,7 +1418,7 @@ mod channels_wire {
use super::*; use super::*;
use serde_json::{json, Value}; use serde_json::{json, Value};
use urus::channels::phoenix::Json; 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<Value>; type P = Json<Value>;
@@ -1603,4 +1603,59 @@ mod channels_wire {
let mut rest = Vec::new(); let mut rest = Vec::new();
let _ = a.read_to_end(&mut rest); 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<P> 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::OnceLock<ChannelHub<P>>> =
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::<LobbySession>(
"room:*",
|_: &str| Box::new(Lobby) as Box<dyn Channel<P>>,
))
});
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");
}
} }