feat(whip): close publishes idle for 30s

crates/server/src/webrtc_ingest.rs

@@ -1,4 +1,4 @@
-use std::{net::SocketAddr, sync::Arc, time::Instant};
+use std::{net::SocketAddr, sync::Arc, time::{Duration, Instant}};
 
 use async_broadcast::broadcast;
 use axum::{
@@ -18,13 +18,16 @@ use str0m::{
     net::{Protocol, Receive},
 };
 use tokio::{net::UdpSocket, sync::mpsc::Receiver};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use crate::{
     StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer,
     http_error::HttpError,
 };
 
+/// Kill a WHIP publish if it delivers no media (video or audio) this long.
+const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30);
+
 pub async fn handle_whip_injest_delete(
     State(state): State<Arc<HttpServer>>,
     ConnectInfo(remote): ConnectInfo<SocketAddr>,
@@ -343,12 +346,12 @@ pub async fn handle_whip_injest(
         .status(StatusCode::CREATED)
         .header(header::CONTENT_TYPE, "application/sdp")
         .header(header::LOCATION, &location)
-        // Provide a STUN server via Link header so OBS can gather
+        // // Provide a STUN server via Link header so OBS can gather
-        // ICE candidates even without explicit STUN configuration.
+        // // ICE candidates even without explicit STUN configuration.
-        .header(
+        // .header(
-            header::LINK,
+        //     header::LINK,
-            "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
+        //     "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
-        )
+        // )
         .body(axum::body::Body::from(answer_sdp))
         .unwrap()
 }
@@ -395,7 +398,7 @@ async fn detach_inject_rtc(
             match rtc.poll_output() {
                 Ok(Output::Timeout(t)) => break t,
                 Ok(Output::Transmit(t)) => {
-                    debug!(
+                    trace!(
                         "Whip TX: {} bytes → {}:{}",
                         t.contents.len(),
                         t.destination.ip(),
@@ -498,13 +501,21 @@ async fn detach_inject_rtc(
                         }
                     }
                 }
-                result = rx.recv() => {
+                    // No UDP input at all for 30s: closing connection.
-                let Some((data, from)) = result else {
+                    // (ICE keepalives still arrive when only the encoder is stalled,
+                    // so this fires on a dead peer, not a paused one.)
+                    result = tokio::time::timeout(NO_MEDIA_TIMEOUT, rx.recv()) => {
+                    let Ok(Some((data, from))) = result else {
+                        if result.is_err() {
+                            warn!(stream_key_id, "Whip: no input for {NO_MEDIA_TIMEOUT:?}, closing connection");
+                            rtc.disconnect();
+                        } else {
                             info!(stream_key_id, "Whip proxy channel closed, cleaning up");
+                        }
                         cleanup().await;
                         return;
                     };
-                debug!(
+                trace!(
                     stream_key_id,
                     len = data.len(),
                     %from,