fix(rtp): disable header extension aggregation on every depayloader

Cargo.toml

@@ -30,6 +30,7 @@ gstreamer = { version = "0.25", features = ["v1_22"] }
 gstreamer-app = { version = "0.25", features = ["v1_22"] }
 gstreamer-controller = { version = "0.25", features = ["v1_22"] }
 gstreamer-net = { version = "0.25", features = ["v1_22"] }
+gstreamer-rtp = { version = "0.25", features = ["v1_22"] }
 gstreamer-gl = { version = "0.25", features = ["v1_22"] }
 gstreamer-video = { version = "0.25", features = ["v1_22"] }
 # gio 0.22 matches the glib 0.22 transitive that gstreamer 0.25 pulls in.

backend/Cargo.toml

@@ -55,6 +55,7 @@ gstreamer-app.workspace = true
 gio.workspace = true
 gstreamer-controller.workspace = true
 gstreamer-net.workspace = true
+gstreamer-rtp.workspace = true
 gstreamer-gl.workspace = true
 gstreamer-video.workspace = true
 gst-plugin-audiofx.workspace = true

backend/src/blocks/builtin/mediaplayer/bridge.rs

@@ -7,6 +7,7 @@
 use super::state::MediaPlayerState;
 use crate::blocks::BlockBuildError;
 use crate::events::EventBroadcaster;
+use crate::gst::rtp_hdrext;
 use gstreamer as gst;
 use gstreamer::prelude::*;
 use gstreamer_app as gst_app;
@@ -123,6 +124,10 @@ pub fn create_decode_pipeline(
         }
     });
 
+    // An `rtsp://` URI makes the source bin autoplug an RTP depayloader, so this
+    // internal pipeline needs the same gstreamer#5057 workaround as the main one.
+    rtp_hdrext::install(&pipeline);
+
     Ok(pipeline)
 }
 
@@ -316,6 +321,10 @@ pub fn create_passthrough_pipeline(
         }
     });
 
+    // An `rtsp://` URI makes the source bin autoplug an RTP depayloader, so this
+    // internal pipeline needs the same gstreamer#5057 workaround as the main one.
+    rtp_hdrext::install(&pipeline);
+
     Ok(pipeline)
 }
 
@@ -602,3 +611,93 @@ pub fn watch_internal_bus(
         bus.remove_signal_watch();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::super::state::Playlist;
+    use super::*;
+    use std::sync::atomic::{AtomicBool, AtomicI64};
+    use std::sync::{Mutex, RwLock};
+
+    /// The bare minimum for the two pipeline constructors: they read `sync` and
+    /// stash a weak ref to the source element, nothing else, before returning.
+    fn test_state() -> Arc<MediaPlayerState> {
+        Arc::new(MediaPlayerState {
+            instance_id: uuid::Uuid::new_v4(),
+            source_element: gst::glib::WeakRef::new(),
+            internal_pipeline: RwLock::new(None),
+            video_appsrc: None,
+            audio_appsrc: None,
+            playlist: RwLock::new(Playlist {
+                files: Vec::new(),
+                current_index: 0,
+            }),
+            is_paused: AtomicBool::new(false),
+            loop_playlist: AtomicBool::new(false),
+            block_id: "test".to_string(),
+            flow_id: uuid::Uuid::new_v4(),
+            switching_file: AtomicBool::new(false),
+            video_linked: AtomicBool::new(false),
+            audio_linked: AtomicBool::new(false),
+            decode: true,
+            sync: true,
+            media_path: std::env::temp_dir(),
+            ts_offset: Arc::new(AtomicI64::new(i64::MIN)),
+            main_pipeline: gst::glib::WeakRef::new(),
+            bus_watch: Mutex::new(None),
+        })
+    }
+
+    /// An `rtsp://` URI makes `uridecodebin`/`urisourcebin` autoplug an RTP
+    /// depayloader inside this pipeline, at which point it needs the
+    /// gstreamer#5057 workaround exactly as much as the main pipeline does —
+    /// the abort takes down the whole process, not just this flow.
+    ///
+    /// CI cannot serve an RTSP stream, so the test stands in for the autoplugged
+    /// element by adding a depayloader to a nested bin after construction. That
+    /// is the same path `deep-element-added` sees, and it fails if the
+    /// `rtp_hdrext::install()` call is removed from the constructor.
+    fn assert_hdrext_disabled_on_late_depayloader(pipeline: &gst::Pipeline) {
+        let inner = gst::Bin::builder().name("inner").build();
+        pipeline.add(&inner).unwrap();
+
+        let depay = gst::ElementFactory::make("rtph264depay")
+            .build()
+            .expect("rtph264depay is in gstreamer1.0-plugins-good, installed in CI")
+            .downcast::<gstreamer_rtp::RTPBaseDepayload>()
+            .expect("rtph264depay derives from GstRTPBaseDepayload");
+        inner.add(&depay).unwrap();
+
+        assert_eq!(
+            rtp_hdrext::is_enabled(&depay),
+            Some(false),
+            "a depayloader autoplugged in the Media Player's internal pipeline still \
+             has RTP header extension aggregation enabled — an interrupted H264 \
+             fragmentation unit from an rtsp:// source will abort the whole process \
+             (gstreamer#5057)"
+        );
+    }
+
+    #[test]
+    fn decode_pipeline_disables_hdrext_aggregation() {
+        let _ = gst::init();
+        if !rtp_hdrext::is_supported() {
+            // GStreamer < 1.24: aggregation does not exist, so neither does the bug.
+            return;
+        }
+        let state = test_state();
+        let pipeline = create_decode_pipeline("test", &state, None).unwrap();
+        assert_hdrext_disabled_on_late_depayloader(&pipeline);
+    }
+
+    #[test]
+    fn passthrough_pipeline_disables_hdrext_aggregation() {
+        let _ = gst::init();
+        if !rtp_hdrext::is_supported() {
+            return;
+        }
+        let state = test_state();
+        let pipeline = create_passthrough_pipeline("test", &state, None).unwrap();
+        assert_hdrext_disabled_on_late_depayloader(&pipeline);
+    }
+}

backend/src/blocks/builtin/whip.rs

@@ -16,6 +16,7 @@ use crate::blocks::{
 };
 use crate::gst::ice_preflight;
 use crate::gst::keyframe_request;
+use crate::gst::rtp_hdrext;
 use crate::whip_session_manager::{SessionCleanupRequest, WhipEndpointConfig};
 use gstreamer as gst;
 use gstreamer::prelude::*;
@@ -1034,6 +1035,11 @@ pub fn create_whipserversrc_for_session(
         .add(&whipserversrc)
         .map_err(|e| format!("Failed to add whipserversrc to session pipeline: {}", e))?;
 
+    // whipserversrc autoplugs RTP depayloaders inside its own bin, so this
+    // pipeline needs the same gstreamer#5057 workaround as the main one.
+    // Install while it is still NULL so no depayloader is missed.
+    rtp_hdrext::install(&session_pipeline);
+
     // Set session pipeline to PLAYING and wait
     session_pipeline
         .set_state(gst::State::Playing)

backend/src/gst/mod.rs

@@ -10,6 +10,7 @@ pub mod ice_preflight;
 pub mod keyframe_request;
 pub mod pipeline;
 pub mod pipeline_monitor;
+pub mod rtp_hdrext;
 pub mod shaders;
 pub mod thread_priority;
 pub mod thumbnail;

backend/src/gst/pipeline/lifecycle.rs

@@ -1,5 +1,5 @@
 use super::{PipelineError, PipelineManager};
-use crate::gst::thread_priority;
+use crate::gst::{rtp_hdrext, thread_priority};
 use gstreamer as gst;
 use gstreamer::prelude::*;
 use strom_types::PipelineState;
@@ -11,6 +11,11 @@ impl PipelineManager {
         info!("Starting pipeline: {}", self.flow_name);
         info!("Pipeline has {} elements", self.elements.len());
 
+        // Disable RTP header extension aggregation before any state changes, so
+        // the handler is in place before decodebin can autoplug a depayloader.
+        // Works around gstreamer#5057, which aborts the whole process.
+        rtp_hdrext::install(&self.pipeline);
+
         // Set up thread priority handler FIRST (before any state changes)
         // This must be done before the pipeline starts so we catch all thread enter events
         info!(