diff --git a/crates/rutster-trunk/Cargo.toml b/crates/rutster-trunk/Cargo.toml index 100047a..cbc99b5 100644 --- a/crates/rutster-trunk/Cargo.toml +++ b/crates/rutster-trunk/Cargo.toml @@ -33,6 +33,7 @@ futures-util = { workspace = true } tokio-tungstenite = { workspace = true } tracing-subscriber = { workspace = true } rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net"] } [features] default = [] diff --git a/crates/rutster-trunk/src/twilio_media_streams.rs b/crates/rutster-trunk/src/twilio_media_streams.rs index 5ba83ca..29f4626 100644 --- a/crates/rutster-trunk/src/twilio_media_streams.rs +++ b/crates/rutster-trunk/src/twilio_media_streams.rs @@ -1,5 +1,7 @@ //! # TwilioMediaStreamsServer -- the inbound WSS pump task. +use std::time::Duration; + use axum::{ Router, extract::{ @@ -28,13 +30,27 @@ pub struct RegisterTrunkInboundChannel { pub reply: oneshot::Sender, } +/// Router state: the media-thread register channel + the app-level +/// keepalive interval (`RUTSTER_WS_PING_SECS`, deploy slice A §5.2). +#[derive(Clone)] +struct MediaStreamState { + register_tx: mpsc::Sender, + ping_interval: Duration, +} + pub struct TwilioMediaStreamsServer; impl TwilioMediaStreamsServer { - pub fn router(register_tx: mpsc::Sender) -> Router { + pub fn router( + register_tx: mpsc::Sender, + ping_interval: Duration, + ) -> Router { Router::new() .route("/twilio/media-stream", get(handle_media_stream)) - .with_state(register_tx) + .with_state(MediaStreamState { + register_tx, + ping_interval, + }) } } @@ -74,14 +90,15 @@ struct OutboundMediaPayload { async fn handle_media_stream( ws: WebSocketUpgrade, - State(register_tx): State>, + State(state): State, ) -> impl IntoResponse { - ws.on_upgrade(move |socket| run_media_stream(socket, register_tx)) + ws.on_upgrade(move |socket| run_media_stream(socket, state.register_tx, state.ping_interval)) } async fn run_media_stream( mut socket: WebSocket, register_tx: mpsc::Sender, + ping_interval: Duration, ) { match socket.recv().await { Some(Ok(Message::Text(txt))) => match serde_json::from_str::(&txt) { @@ -150,6 +167,17 @@ async fn run_media_stream( }; info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active"); + // App-level keepalive (deploy slice A §5.2): engine-originated pings + // at RUTSTER_WS_PING_SECS cadence. `interval_at` skips the immediate + // first tick (a fresh call doesn't need a ping at t=0); + // MissedTickBehavior::Delay because a late ping must not be followed + // by a compensating burst. The pre-registration handshake recvs above + // are un-pinged — Twilio sends connected+start immediately on + // connect, so that window is sub-second by protocol. + let mut ping = + tokio::time::interval_at(tokio::time::Instant::now() + ping_interval, ping_interval); + ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { tokio::select! { biased; @@ -220,6 +248,15 @@ async fn run_media_stream( break; } }, + + _ = ping.tick() => { + // Vec::new(): an empty application payload — the frame + // itself is the keepalive; peers must Pong per RFC 6455. + if socket.send(Message::Ping(Vec::new())).await.is_err() { + warn!(%call_sid, "twilio WS ping send failed; ending pump"); + break; + } + }, } } @@ -342,7 +379,7 @@ mod tests { #[tokio::test] async fn router_constructs_with_register_channel() { let (register_tx, _register_rx) = mpsc::channel::(4); - let _app = TwilioMediaStreamsServer::router(register_tx); + let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20)); } fn rms(samples: &[i16]) -> f64 { diff --git a/crates/rutster-trunk/tests/ws_ping.rs b/crates/rutster-trunk/tests/ws_ping.rs new file mode 100644 index 0000000..b674b59 --- /dev/null +++ b/crates/rutster-trunk/tests/ws_ping.rs @@ -0,0 +1,69 @@ +//! App-level WS ping emission on the trunk media-stream WS (deploy +//! slice A §5.2). +//! +//! Spins the real router on an ephemeral listener, performs the Twilio +//! connected/start handshake as a tokio-tungstenite client, then asserts +//! an engine-originated Ping frame arrives well within CI slack of the +//! configured interval. Twilio documents no keepalive of its own — this +//! test is the contract that WE originate the pings. + +use futures_util::{SinkExt, StreamExt}; +use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer}; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn engine_originates_ws_ping_at_configured_interval() { + let (register_tx, mut register_rx) = mpsc::channel::(4); + let app = TwilioMediaStreamsServer::router(register_tx, std::time::Duration::from_millis(100)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + // Stub media thread: ack the registration and hold the channel ends + // alive so the pump loop stays up for the whole test. + let registered = tokio::spawn(async move { + let register = register_rx.recv().await.expect("pump registers"); + let RegisterTrunkInboundChannel { + inbound_from_twilio_rx, + outbound_to_twilio_tx, + reply, + .. + } = register; + let _ = reply.send(rutster_call_model::ChannelId::new()); + (inbound_from_twilio_rx, outbound_to_twilio_tx) + }); + + let (mut ws, _resp) = + tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream")) + .await + .expect("WS connect"); + ws.send(Message::Text( + r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(), + )) + .await + .unwrap(); + ws.send(Message::Text( + r#"{"event":"start","start":{"streamSid":"MZping","callSid":"CAping"}}"#.into(), + )) + .await + .unwrap(); + let _register = registered.await.unwrap(); + + // 100 ms interval → a Ping must land within 2 s (20 intervals of + // slack for a loaded CI runner; the point is emission, not cadence + // precision). + let got_ping = tokio::time::timeout(std::time::Duration::from_secs(2), async { + while let Some(msg) = ws.next().await { + if matches!(msg.expect("WS frame"), Message::Ping(_)) { + return true; + } + } + false + }) + .await + .expect("a Ping frame must arrive before the deadline"); + assert!(got_ping, "expected an engine-originated Ping frame"); +} diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs index 3dc7589..44f1b43 100644 --- a/crates/rutster/src/config.rs +++ b/crates/rutster/src/config.rs @@ -95,6 +95,36 @@ pub fn drain_deadline(raw: Option) -> Result) -> Result { + match raw { + None => Ok(std::time::Duration::from_secs(20)), + Some(s) => { + let secs: u64 = s + .parse() + .map_err(|e| format!("RUTSTER_WS_PING_SECS {s:?}: {e}"))?; + if secs == 0 { + return Err( + "RUTSTER_WS_PING_SECS must be >= 1 (0 would busy-loop the ping timer)" + .to_string(), + ); + } + Ok(std::time::Duration::from_secs(secs)) + } + } +} + /// Parse the four `RUTSTER_TWILIO_*` env vars into a [`TwilioCredentials`]. /// /// Returns `Ok(None)` when ALL four vars are unset — the binary runs WebRTC-only @@ -316,4 +346,32 @@ mod tests { .unwrap_err(); assert!(err.contains("RUTSTER_TWILIO_WEBHOOK_BASE")); } + + #[test] + fn ws_ping_interval_defaults_to_20s_when_unset() { + assert_eq!( + ws_ping_interval(None).unwrap(), + std::time::Duration::from_secs(20) + ); + } + + #[test] + fn ws_ping_interval_parses_override() { + assert_eq!( + ws_ping_interval(Some("5".into())).unwrap(), + std::time::Duration::from_secs(5) + ); + } + + #[test] + fn ws_ping_interval_rejects_zero() { + let err = ws_ping_interval(Some("0".into())).unwrap_err(); + assert!(err.contains("RUTSTER_WS_PING_SECS")); + } + + #[test] + fn ws_ping_interval_rejects_garbage_with_var_name_in_error() { + let err = ws_ping_interval(Some("often".into())).unwrap_err(); + assert!(err.contains("RUTSTER_WS_PING_SECS")); + } } diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index 4804bf4..a8cf3bb 100644 --- a/crates/rutster/src/main.rs +++ b/crates/rutster/src/main.rs @@ -128,8 +128,15 @@ async fn main() { } } }); - let trunk_router = - rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(trunk_register_tx); + // Deploy slice A §5.2: app-level keepalive on the only long-lived WS + // surface. Fail-fast on garbage, like every RUTSTER_* knob. + let ws_ping_interval = + rutster::config::ws_ping_interval(std::env::var("RUTSTER_WS_PING_SECS").ok()) + .expect("RUTSTER_WS_PING_SECS must be a positive integer (seconds)"); + let trunk_router = rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router( + trunk_register_tx, + ws_ping_interval, + ); let app = router(app_state).merge(trunk_router);