diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index 84f7913..c659495 100644 --- a/crates/rutster/src/media_thread.rs +++ b/crates/rutster/src/media_thread.rs @@ -285,21 +285,11 @@ fn run_media_thread( } 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)"); - } - } + if let Some(conn) = s.tap_conn.take() { + // Hand the 750ms bounded teardown to tokio — + // the tick loop never waits on brain I/O + // (review M7; see spawn_tap_teardown docs). + crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn); } s.rtc.channel.tap = None; s.rtc.channel.state = rutster_call_model::ChannelState::Closed; diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index 94e30fc..6fcb831 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -111,6 +111,33 @@ pub struct TapConn { pub tool_registry: Arc>, } +/// Tear down one session's tap engine WITHOUT blocking the caller. +/// +/// # Why this must never run inline on the media thread (review M7) +/// +/// The old Delete arm did `tokio_handle.block_on(timeout(750ms, join))` +/// inside the tick loop: one teardown with an unresponsive brain froze +/// the 20 ms loop for EVERY live call on the node (~37 missed frames), +/// and batched Deletes stacked sequentially. Teardown is brain I/O — +/// exactly what the tick loop must never wait on. Same discipline as the +/// tap pipe's try_send posture. +pub fn spawn_tap_teardown(tokio_handle: &tokio::runtime::Handle, id: ChannelId, mut conn: TapConn) { + tokio_handle.spawn(async move { + // Gentle path first (AGENTS.md: close_tx is the documented + // trigger, abort is the safety net). + let _ = conn.close_tx.send(()); + match tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await { + Ok(Ok(())) => { + info!(channel_id = %id, "tap engine torn down (graceful)"); + } + _ => { + conn.join.abort(); + info!(channel_id = %id, "tap engine torn down (abort after timeout)"); + } + } + }); +} + /// Spawn the TapEngine task for one session. Dials `tap_url`, runs the pump /// loop, reconnects with bounded backoff on failure (spec §4.3, §5.2). /// @@ -456,6 +483,47 @@ mod tests { use super::*; use rutster_media::AudioSource; + #[test] + fn spawn_tap_teardown_returns_immediately_and_aborts_stuck_engine() { + let rt = tokio::runtime::Runtime::new().unwrap(); + // A "stuck brain": the engine task never finishes on its own. Its + // guard's Drop fires on abort — our proof the teardown completed. + struct DropGuard(std::sync::mpsc::Sender<()>); + impl Drop for DropGuard { + fn drop(&mut self) { + let _ = self.0.send(()); + } + } + let (dropped_tx, dropped_rx) = std::sync::mpsc::channel(); + let join = rt.spawn(async move { + let _guard = DropGuard(dropped_tx); + std::future::pending::<()>().await; + }); + let (close_tx, _close_rx) = oneshot::channel(); + let conn = TapConn { + close_tx, + join, + metrics: TapMetrics::new(), + flush_rx: None, + rx_function_call: None, + tx_function_call_output: None, + tool_registry: Arc::new(Mutex::new(ToolRegistry::default())), + }; + + let started = std::time::Instant::now(); + spawn_tap_teardown(rt.handle(), rutster_call_model::ChannelId::new(), conn); + // The whole point (review M7): the caller — the media TICK LOOP — + // must not wait out the 750ms brain timeout. + assert!( + started.elapsed() < Duration::from_millis(100), + "teardown must not block the caller" + ); + // …but the stuck engine task must still get aborted after the cap. + dropped_rx + .recv_timeout(Duration::from_secs(2)) + .expect("stuck engine task was aborted (guard dropped)"); + } + #[test] fn backoff_doubles_until_cap() { let mut b = Backoff::default();