diff --git a/crates/server/src/webrtc_ingest.rs b/crates/server/src/webrtc_ingest.rs index 9ee1a2a..ba5cc6e 100644 --- a/crates/server/src/webrtc_ingest.rs +++ b/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>, ConnectInfo(remote): ConnectInfo, @@ -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 - // ICE candidates even without explicit STUN configuration. - .header( - header::LINK, - "; rel=\"ice-server\"", - ) + // // Provide a STUN server via Link header so OBS can gather + // // ICE candidates even without explicit STUN configuration. + // .header( + // header::LINK, + // "; 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() => { - let Some((data, from)) = result else { - info!(stream_key_id, "Whip proxy channel closed, cleaning up"); - cleanup().await; - return; - }; - debug!( + // No UDP input at all for 30s: closing connection. + // (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; + }; + trace!( stream_key_id, len = data.len(), %from,