feat(trunk): app-level WS ping on the media-stream WS, RUTSTER_WS_PING_SECS default 20 (deploy-A §5.2)

Neither Twilio nor Telnyx documents WS keepalive — keepalive is entirely
our job (spec §3.1 invariant 3), and the caller→engine direction can be
the only traffic for hours, letting any 60s proxy idle timer kill a live
call. Engine-originated Ping frames as a third select! branch in the
existing pump loop; interval via the config.rs fail-fast parser pattern
(0 rejected — no 'off' spelling by design).

Router signature grows the interval; both call sites updated. WebRTC
signaling is plain HTTP today, so this is the only long-lived WS.

Signed-off-by: Aaron D. Lee <himself@adlee.work>

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 = []

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<ChannelId>,
 }
 
+/// 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<RegisterTrunkInboundChannel>,
+    ping_interval: Duration,
+}
+
 pub struct TwilioMediaStreamsServer;
 
 impl TwilioMediaStreamsServer {
-    pub fn router(register_tx: mpsc::Sender<RegisterTrunkInboundChannel>) -> Router {
+    pub fn router(
+        register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
+        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<mpsc::Sender<RegisterTrunkInboundChannel>>,
+    State(state): State<MediaStreamState>,
 ) -> 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<RegisterTrunkInboundChannel>,
+    ping_interval: Duration,
 ) {
     match socket.recv().await {
         Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&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::<RegisterTrunkInboundChannel>(4);
-        let _app = TwilioMediaStreamsServer::router(register_tx);
+        let _app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
     }
 
     fn rms(samples: &[i16]) -> f64 {

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::<RegisterTrunkInboundChannel>(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");
+}

crates/rutster/src/config.rs

@@ -95,6 +95,36 @@ pub fn drain_deadline(raw: Option<String>) -> Result<std::time::Duration, String
     }
 }
 
+/// `RUTSTER_WS_PING_SECS` — engine-originated app-level WS ping interval
+/// on long-lived WebSockets. Today that is exactly one surface: the trunk
+/// media-stream WS (WebRTC signaling is plain HTTP). Default 20 s.
+///
+/// Belt-and-braces against every idle timer between the FOB and the
+/// CPaaS: neither Twilio nor Telnyx documents WS keepalive of their own,
+/// so keepalive is entirely our job (deploy spec §3.1 invariant 3), and
+/// the caller→engine direction may be the only traffic for hours.
+///
+/// `0` is rejected fail-fast: it would busy-loop the ping timer. There is
+/// deliberately no "off" spelling — an operator who wants no pings is an
+/// operator about to learn about 60 s proxy idle defaults mid-call.
+pub fn ws_ping_interval(raw: Option<String>) -> Result<std::time::Duration, String> {
+    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"));
+    }
 }

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);