crates/rutster/src/lib.rs

@@ -22,6 +22,7 @@
 //! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md)
 //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical.
 
+pub mod media_thread;
 pub mod routes;
 pub mod session_map;
 pub mod tap_engine;

crates/rutster/src/media_thread.rs

@@ -0,0 +1,308 @@
+//! # MediaThread — the dedicated 20ms media loop on a std::thread
+//! (slice-4 spec §2.2, §4)
+//!
+//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
+//! tokio pool." slice-1 ran the poll on tokio as an acknowledged
+//! deviation; slice-4 graduates it. ONE `std::thread::spawn` at binary
+//! startup owns `HashMap<ChannelId, RtcSession>` exclusively; all access
+//! from axum is via a command channel. The 10ms meta-tick between loop
+//! iterations is `std::thread::sleep(Duration::from_millis(10))`.
+//!
+//! # Why one thread, not per-session
+//!
+//! Spearhead scale (see slice-4 spec §6.3). The command-channel seam
+//! makes the later threadpool-shard graduation localized.
+//!
+//! # The seam (loop_driver + rtc_session byte-identical)
+//!
+//! `MediaThread` calls `RtcSession::run_poll_once(now)` — the unchanged
+//! `loop_driver::drive`. The `Reflex<TapAudioPipe>` wrapper is wired in
+//! here on the `Connected` transition (via `RtcSession::set_pipe`), not
+//! inside `rtc_session.rs`. The seam holds.
+
+use std::collections::HashMap;
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use rutster_call_model::ChannelId;
+use rutster_media::RtcSession;
+use tokio::sync::{mpsc, oneshot};
+use tracing::{debug, info};
+
+use crate::tap_engine::spawn_tap_engine;
+
+/// The 10ms meta-tick. Finer than the 20ms outbound encode tick so str0m's
+/// `Timeout` outputs are honored promptly.
+const META_TICK: Duration = Duration::from_millis(10);
+
+/// Capacity for the command channel from axum to the media thread.
+const CMD_CHANNEL_CAPACITY: usize = 64;
+
+/// Commands axum sends to the media thread (cold-path only — NEVER on
+/// the 20ms tick). The thread owns RtcSessions exclusively; this is the
+/// ONLY entry point for axum-side mutation.
+#[derive(Debug)]
+pub enum MediaCmd {
+    /// Construct a fresh RtcSession, store it under a new ChannelId, reply.
+    /// The thread constructs RtcSession::new() (keeps all RtcSession
+    /// construction on the thread that owns it).
+    Register {
+        tap_url: url::Url,
+        reply: oneshot::Sender<Result<ChannelId, String>>,
+    },
+    /// Accept a browser SDP offer on the session's behalf, reply with the
+    /// SDP answer (cold-path — the axum POST /v1/sessions/{id}/office handler).
+    AcceptOffer {
+        id: ChannelId,
+        sdp: String,
+        reply: oneshot::Sender<Result<String, String>>,
+    },
+    /// Tear down a session — fires close_tx + bounded-await the engine task
+    /// (750ms cap), then removes the entry.
+    Delete {
+        id: ChannelId,
+        reply: oneshot::Sender<()>,
+    },
+    /// Graceful shutdown — drain + drop + join.
+    Shutdown { reply: oneshot::Sender<()> },
+}
+
+/// The handle returned to the binary. Clone the `cmd_tx` per-session;
+/// the `JoinHandle` is dropped on shutdown to detach (the binary's
+/// graceful-shutdown signal fires `Shutdown` first).
+pub struct MediaThread {
+    cmd_tx: mpsc::Sender<MediaCmd>,
+    join: Option<JoinHandle<()>>,
+}
+
+impl MediaThread {
+    /// Return a clone of the command-channel sender, so the binary (axum
+    /// routes / `AppState`) can issue cold-path commands to the thread.
+    pub fn cmd_tx(&self) -> mpsc::Sender<MediaCmd> {
+        self.cmd_tx.clone()
+    }
+
+    /// Spawn the dedicated media thread. Captures a `tokio::runtime::Handle`
+    /// so the thread can `handle.spawn(spawn_tap_engine(...))` on the
+    /// `Connected` transition. The thread owns `HashMap<ChannelId,
+    /// ThreadSession>` exclusively.
+    ///
+    /// Returns `std::io::Error` if the OS cannot create the std::thread
+    /// (for example: the process has hit its thread limit). Thread creation
+    /// is part of the cold startup path, not the 20ms media loop, so the
+    /// error is propagated for the binary to decide whether to abort, retry,
+    /// or degrade. Inside the loop we still follow the "drop + observe, don't
+    /// crash" policy.
+    pub fn spawn(
+        default_tap_url: url::Url,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result<Self, std::io::Error> {
+        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
+        let join = std::thread::Builder::new()
+            .name("rutster-media".into())
+            .spawn(move || {
+                run_media_thread(cmd_rx, default_tap_url, tokio_handle);
+            })?;
+        Ok(Self {
+            cmd_tx,
+            join: Some(join),
+        })
+    }
+
+    /// Graceful shutdown — drains commands + joins the thread.
+    pub fn shutdown(mut self) {
+        let (reply, rx) = oneshot::channel();
+        let _ = self.cmd_tx.blocking_send(MediaCmd::Shutdown { reply });
+        let _ = rx.blocking_recv();
+        if let Some(join) = self.join.take() {
+            let _ = join.join();
+        }
+    }
+}
+
+impl Drop for MediaThread {
+    fn drop(&mut self) {
+        if let Some(join) = self.join.take() {
+            // Best-effort: if shutdown wasn't called explicitly, just
+            // detach. The thread will exit when cmd_rx is dropped.
+            // (We don't block on join in Drop — that could deadlock
+            // if the thread is mid-call into a tokio runtime handle
+            // that's being torn down.)
+            debug!(name = ?join.thread().name(), "media thread detached on drop");
+        }
+    }
+}
+
+/// The per-session state owned by the media thread.
+struct ThreadSession {
+    rtc: RtcSession,
+    /// `Some` only after the `Connected` transition spawns the TapEngine
+    /// + wires the `Reflex<TapAudioPipe>` wrapper.
+    tap_conn: Option<crate::tap_engine::TapConn>,
+}
+
+fn run_media_thread(
+    mut cmd_rx: mpsc::Receiver<MediaCmd>,
+    default_tap_url: url::Url,
+    tokio_handle: tokio::runtime::Handle,
+) {
+    let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
+    info!("media thread started");
+
+    loop {
+        // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. ===
+        while let Ok(cmd) = cmd_rx.try_recv() {
+            match cmd {
+                MediaCmd::Register { tap_url, reply } => match RtcSession::new() {
+                    Ok(session) => {
+                        let id = session.channel_id();
+                        sessions.insert(
+                            id,
+                            ThreadSession {
+                                rtc: session,
+                                tap_conn: None,
+                            },
+                        );
+                        let _ = reply.send(Ok(id));
+                        debug!(channel_id = %id, %tap_url, "session registered");
+                    }
+                    Err(e) => {
+                        let _ = reply.send(Err(format!("RtcSession::new: {e}")));
+                    }
+                },
+                MediaCmd::AcceptOffer { id, sdp, reply } => {
+                    let result = match sessions.get_mut(&id) {
+                        Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")),
+                        None => Err(format!("session {id} not found")),
+                    };
+                    let _ = reply.send(result);
+                }
+                MediaCmd::Delete { id, reply } => {
+                    if let Some(mut s) = sessions.remove(&id) {
+                        if let Some(mut conn) = s.tap_conn.take() {
+                            let _ = conn.close_tx.send(());
+                            let teardown = tokio_handle.block_on(tokio::time::timeout(
+                                Duration::from_millis(750),
+                                &mut conn.join,
+                            ));
+                            match teardown {
+                                Ok(Ok(())) => {
+                                    info!(channel_id = %id, "tap engine torn down via Delete (graceful)");
+                                }
+                                _ => {
+                                    conn.join.abort();
+                                    info!(channel_id = %id, "tap engine torn down via Delete (abort after timeout)");
+                                }
+                            }
+                        }
+                        s.rtc.channel.tap = None;
+                        s.rtc.channel.state = rutster_call_model::ChannelState::Closed;
+                    }
+                    let _ = reply.send(());
+                }
+                MediaCmd::Shutdown { reply } => {
+                    info!(
+                        "media thread shutdown; dropping {} sessions",
+                        sessions.len()
+                    );
+                    sessions.clear();
+                    let _ = reply.send(());
+                    return;
+                }
+            }
+        }
+
+        // === Step 2: the 10ms meta-tick over all sessions. ===
+        let now = Instant::now();
+        let mut closed_ids: Vec<ChannelId> = Vec::new();
+        for (id, session) in sessions.iter_mut() {
+            // Drain flush side-channel BEFORE run_poll_once (slice-2 §5.3 step 4).
+            if let Some(conn) = session.tap_conn.as_mut() {
+                if let Some(rx) = conn.flush_rx.as_mut() {
+                    let mut should_flush = false;
+                    while let Ok(()) = rx.try_recv() {
+                        should_flush = true;
+                    }
+                    if should_flush {
+                        session.rtc.clear_playout_ring();
+                    }
+                }
+            }
+            let _ = session.rtc.run_poll_once(now);
+
+            // === relay the Connected transition: spawn TapEngine + wire Reflex. ===
+            use rutster_call_model::ChannelState;
+            if let ChannelState::Connected = session.rtc.channel.state {
+                if session.rtc.channel.tap.is_none() {
+                    let url = default_tap_url.clone();
+                    // The media thread owns the advisory channel (multi-producer:
+                    // tokio::sync::mpsc::Sender is Clone). Both the brain's
+                    // advisories (via spawn_tap_engine) AND the local VAD's
+                    // trips (via LocalVadReflex) push to the SAME mpsc; the
+                    // Reflex drains both uniformly. This means spawn_tap_engine
+                    // takes advisory_tx as a PARAMETER (Task 5's signature
+                    // changes: spawn_tap_engine(session_id, tap_url, app_state,
+                    // advisory_tx) — see Task 5's revision note).
+                    let (advisory_tx, advisory_rx) =
+                        mpsc::channel::<rutster_media::AdvisoryEvent>(16);
+                    let (pipe, conn) = tokio_handle.block_on(async {
+                        spawn_tap_engine(
+                            *id,
+                            url,
+                            crate::session_map::AppState::default(),
+                            advisory_tx.clone(),
+                        )
+                    });
+                    let metrics = rutster_media::ReflexMetrics::new();
+                    // Compose: Reflex<TapAudioPipe> (state machine) wrapped by
+                    // LocalVadReflex (primary VAD trigger). Both feed advisory_tx.
+                    let reflex = rutster_media::Reflex::new(pipe, advisory_rx, metrics);
+                    let vad = rutster_media::LocalVadReflex::new(reflex, advisory_tx);
+                    session.rtc.set_pipe(vad);
+                    session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
+                    session.tap_conn = Some(conn);
+                    info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
+                    continue;
+                }
+            }
+
+            if session.rtc.is_closed() {
+                closed_ids.push(*id);
+            }
+        }
+        for id in closed_ids {
+            sessions.remove(&id);
+            debug!(channel_id = %id, "session evicted after close");
+        }
+
+        // === Step 3: sleep META_TICK. ===
+        std::thread::sleep(META_TICK);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn media_thread_register_and_shutdown_round_trips() {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let thread = MediaThread::spawn(url, handle).expect("media thread spawn in test");
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let id = rx
+            .blocking_recv()
+            .expect("register reply")
+            .expect("session");
+        assert_eq!(format!("{}", id).len(), 36, "UUID-shaped ChannelId");
+        thread.shutdown();
+    }
+}