Skip to content
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 63 additions & 3 deletions backend/src/blocks/builtin/whep.rs
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,17 @@ fn explicit_track_count(properties: &HashMap<String, PropertyValue>, name: &str)
})
}

/// Parse do_retransmission from properties (default: true).
fn parse_do_retransmission(properties: &HashMap<String, PropertyValue>) -> bool {
properties
.get("do_retransmission")
.and_then(|v| match v {
PropertyValue::Bool(b) => Some(*b),
_ => None,
})
.unwrap_or(true)
}

/// Migrate a legacy `mode` property on a WHEP Output block to explicit
/// `num_audio_tracks` / `num_video_tracks` counts, and drop `mode`.
///
Expand Down Expand Up @@ -951,6 +962,8 @@ fn build_whepserversink(
num_audio_tracks, num_video_tracks
);

let do_retransmission = parse_do_retransmission(properties);

// Timestamp offset in milliseconds. A negative value shifts playout earlier,
// reducing end-to-end latency for this output while maintaining A/V sync.
// Applied as ts-offset on clocksync and appsink inside whepserversink.
Expand Down Expand Up @@ -1021,16 +1034,19 @@ fn build_whepserversink(
whepserversink.set_property("turn-servers", turn_servers);
}

// Disable FEC but keep RTX (retransmission) enabled.
// Disable FEC but keep RTX (retransmission) configurable (default: on).
// - FEC adds proactive redundancy packets on every stream (~50% constant
// overhead, near-double bandwidth for pre-encoded high-bitrate video),
// so it stays off.
// - RTX is reactive: it costs nothing while no packets are lost and only
// resends the exact packets the client NACKs. Without it, every loss
// escalates to PLI -> forced keyframe, which is far more expensive and
// leaves the picture broken until the keyframe arrives.
// leaves the picture broken until the keyframe arrives. It can be
// disabled as a workaround for a GStreamer bug where RTX combined with
// RTP header-extension aggregation can crash the process (assertion in
// gst_rtp_base_depayload_handle_buffer on priv->hdrext_buffers).
whepserversink.set_property("do-fec", false);
whepserversink.set_property("do-retransmission", true);
whepserversink.set_property("do-retransmission", do_retransmission);

// Access the signaller child and set its properties
// Bind to localhost only - axum will proxy external requests
Expand Down Expand Up @@ -2341,6 +2357,20 @@ fn whep_output_definition() -> BlockDefinition {
live: false,
persist: None,
},
ExposedProperty {
name: "do_retransmission".to_string(),
label: "Retransmission (RTX)".to_string(),
description: "Resend lost packets to viewers on request (NACK-based). Disable only for diagnostics — a known GStreamer bug can crash the process when RTX is combined with RTP header-extension aggregation; without it, packet loss forces a full keyframe request instead of a cheap resend.".to_string(),
property_type: PropertyType::Bool,
default_value: Some(PropertyValue::Bool(true)),
mapping: PropertyMapping {
element_id: "_block".to_string(),
property_name: "do_retransmission".to_string(),
transform: None,
},
live: false,
persist: None,
},
],
// Note: external_pads here are the static defaults (1 video + 1 audio).
// The actual pads are determined dynamically by WHEPOutputBuilder::get_external_pads()
Expand Down Expand Up @@ -2416,6 +2446,36 @@ mod tests {
.collect()
}

/// Build a property map from explicit key/value pairs, for tests of pure
/// parsing helpers that don't need the pad-count `props` helper above.
fn raw_props(entries: &[(&str, PropertyValue)]) -> HashMap<String, PropertyValue> {
entries
.iter()
.map(|(k, v)| (k.to_string(), v.clone()))
.collect()
}

#[test]
fn do_retransmission_defaults_to_true() {
assert!(parse_do_retransmission(&raw_props(&[])));
}

#[test]
fn do_retransmission_respects_explicit_true() {
assert!(parse_do_retransmission(&raw_props(&[(
"do_retransmission",
PropertyValue::Bool(true)
)])));
}

#[test]
fn do_retransmission_respects_explicit_false() {
assert!(!parse_do_retransmission(&raw_props(&[(
"do_retransmission",
PropertyValue::Bool(false)
)])));
}

#[test]
fn external_pads_default_is_one_video_and_one_audio() {
let pads = WHEPOutputBuilder
Expand Down
67 changes: 65 additions & 2 deletions backend/src/blocks/builtin/whip.rs
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,17 @@ impl BlockBuilder for WHIPInputBuilder {
// WHIP Input (whipserversrc - hosts WHIP server)
// ============================================================================

/// Parse do_retransmission from properties (default: true).
fn parse_do_retransmission(properties: &HashMap<String, PropertyValue>) -> bool {
properties
.get("do_retransmission")
.and_then(|v| match v {
PropertyValue::Bool(b) => Some(*b),
_ => None,
})
.unwrap_or(true)
}

/// Build WHIP Input per-slot output chains.
///
/// At build time, per-slot chains are created in the main pipeline:
Expand Down Expand Up @@ -181,6 +192,8 @@ fn build_whipserversrc(
})
.unwrap_or(true);

let do_retransmission = parse_do_retransmission(properties);

let max_video_bitrate_kbps = properties
.get("max_video_bitrate")
.and_then(|v| match v {
Expand Down Expand Up @@ -415,8 +428,8 @@ fn build_whipserversrc(
let turn_server = ctx.turn_server();

info!(
"WHIP Input configured: endpoint_id='{}', stun={:?}, turn={:?}, mode={:?}, decode={}, max_sessions={} (whipserversrc created per-session)",
endpoint_id, stun_server, turn_server, mode, decode, max_sessions
"WHIP Input configured: endpoint_id='{}', stun={:?}, turn={:?}, mode={:?}, decode={}, do_retransmission={}, max_sessions={} (whipserversrc created per-session)",
endpoint_id, stun_server, turn_server, mode, decode, do_retransmission, max_sessions
);

// Register WHIP endpoint with the build context (port=0 placeholder, sessions get their own ports)
Expand All @@ -436,6 +449,7 @@ fn build_whipserversrc(
ice_transport_policy: ctx.ice_transport_policy().to_string(),
pipeline_weak: gst::glib::WeakRef::new(),
decode,
do_retransmission,
dynamic_webrtcbin_store: ctx.dynamic_webrtcbin_store(),
max_video_bitrate_kbps,
max_sessions,
Expand Down Expand Up @@ -511,6 +525,8 @@ pub fn create_whipserversrc_for_session(
let signaller = whipserversrc.property::<gst::glib::Object>("signaller");
signaller.set_property("host-addr", &host_addr);

whipserversrc.set_property("do-retransmission", config.do_retransmission);

// Configure codec negotiation based on mode
if config.mode.has_audio() {
let audio_codecs = gst::Array::new(["OPUS"]);
Expand Down Expand Up @@ -1444,6 +1460,20 @@ fn whip_input_definition() -> BlockDefinition {
live: false,
persist: None,
},
ExposedProperty {
name: "do_retransmission".to_string(),
label: "Retransmission (RTX)".to_string(),
description: "Request retransmission of lost packets from the publisher (NACK-based). Disable only for diagnostics; without it, any packet loss forces a full keyframe request instead of a cheap resend.".to_string(),
property_type: PropertyType::Bool,
default_value: Some(PropertyValue::Bool(true)),
mapping: PropertyMapping {
element_id: "_block".to_string(),
property_name: "do_retransmission".to_string(),
transform: None,
},
live: false,
persist: None,
},
ExposedProperty {
name: "max_video_bitrate".to_string(),
label: "Max Video Bitrate (kbps)".to_string(),
Expand Down Expand Up @@ -1625,3 +1655,36 @@ fn setup_incoming_rtp_handler(

info!("WHIP: Incoming RTP handler installed for {}", instance_id);
}

#[cfg(test)]
mod tests {
use super::*;

fn props(entries: &[(&str, PropertyValue)]) -> HashMap<String, PropertyValue> {
entries
.iter()
.map(|(k, v)| (k.to_string(), v.clone()))
.collect()
}

#[test]
fn do_retransmission_defaults_to_true() {
assert!(parse_do_retransmission(&props(&[])));
}

#[test]
fn do_retransmission_respects_explicit_true() {
assert!(parse_do_retransmission(&props(&[(
"do_retransmission",
PropertyValue::Bool(true)
)])));
}

#[test]
fn do_retransmission_respects_explicit_false() {
assert!(!parse_do_retransmission(&props(&[(
"do_retransmission",
PropertyValue::Bool(false)
)])));
}
}
3 changes: 3 additions & 0 deletions backend/src/whip_session_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ pub struct WhipEndpointConfig {
pub pipeline_weak: gst::glib::WeakRef<gst::Pipeline>,
/// Whether to decode RTP to raw media (true) or pass through RTP (false)
pub decode: bool,
/// Whether whipserversrc should request retransmission (NACK) of lost
/// packets from the publisher. Matches upstream default (true).
pub do_retransmission: bool,
/// Shared dynamic webrtcbin store for ICE policy tracking
pub dynamic_webrtcbin_store: DynamicWebrtcbinStore,
/// Maximum video bitrate hint for Chrome (kbps). Injected into the SDP
Expand Down
Loading