From c7db33a8d2c1fc1a3281f70af7010a9d48ac7770 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Thu, 27 Aug 2026 16:13:09 -0700 Subject: [PATCH 1/2] fix(rtp): disable header extension aggregation on every depayloader Since 1.24 GstRTPBaseDepayload caches the RTP header of every packet feeding the output buffer it is assembling, clearing the cache only when the subclass pushes or flushes. gst_rtp_base_depayload_delayed() means "this packet's header belongs to the next output buffer", and the base class asserts the cache is empty when that happens. rtph264depay breaks that invariant: an interrupted fragmentation unit calls delayed() and then finish_fragmentation_unit(), which in access-unit mode can absorb the truncated NAL without producing an output buffer. Nothing is pushed, the cache is still populated, and gstrtpbasedepayload.c:942:gst_rtp_base_depayload_handle_buffer: 'gst_buffer_list_length (priv->hdrext_buffers) == 0' should be TRUE aborts the process, taking every unrelated flow on the server with it. g_assert_true is not defusable, so the only fix available to us is to stop the cache from filling. Turning aggregation off restores the pre-1.24 behaviour: header extensions are read from the current packet instead of accumulated. Strom reads no header-extension metadata, and the extensions that matter for transport (transport-cc, abs-send-time, mid) are consumed by webrtcbin well upstream of any depayloader, so nothing is lost. There is no GObject property for this - the C setter is the only switch, and it is Since: 1.24. Binding it normally would raise the workspace build floor from 1.22, breaking the default install on Debian 12 and the default ARM64 cross-compile target (Raspberry Pi OS 12), both of which ship GStreamer 1.22. Resolving the symbol at runtime keeps one binary working on both: below 1.24 the lookup fails and we do nothing, which is correct because aggregation - and therefore the bug - does not exist there. install() sweeps elements already in the pipeline and connects deep-element-added for ones decodebin autoplugs later. It runs at the top of start(), before any state change, so no depayloader is missed. The handler captures nothing, so it creates no reference cycle. Refs: https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/5057 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01E2xUTyh7HkjawtC8CxMwh6 --- Cargo.lock | 1 + Cargo.toml | 1 + backend/Cargo.toml | 1 + backend/src/gst/mod.rs | 1 + backend/src/gst/pipeline/lifecycle.rs | 7 +- backend/src/gst/rtp_hdrext.rs | 271 +++++++++++++++++++ backend/tests/rtp_hdrext_aggregation_test.rs | 116 ++++++++ 7 files changed, 397 insertions(+), 1 deletion(-) create mode 100644 backend/src/gst/rtp_hdrext.rs create mode 100644 backend/tests/rtp_hdrext_aggregation_test.rs diff --git a/Cargo.lock b/Cargo.lock index cf34c8b8..41e7ca91 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6607,6 +6607,7 @@ dependencies = [ "gstreamer-controller", "gstreamer-gl", "gstreamer-net", + "gstreamer-rtp", "gstreamer-video", "hostname", "http-body-util", diff --git a/Cargo.toml b/Cargo.toml index 88144835..8372582a 100644 --- a/Cargo.toml +++ b/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. diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 4bc4d2e6..df634b50 100644 --- a/backend/Cargo.toml +++ b/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 diff --git a/backend/src/gst/mod.rs b/backend/src/gst/mod.rs index 18b954e7..ea392311 100644 --- a/backend/src/gst/mod.rs +++ b/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; diff --git a/backend/src/gst/pipeline/lifecycle.rs b/backend/src/gst/pipeline/lifecycle.rs index 728c7590..6959341e 100644 --- a/backend/src/gst/pipeline/lifecycle.rs +++ b/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!( diff --git a/backend/src/gst/rtp_hdrext.rs b/backend/src/gst/rtp_hdrext.rs new file mode 100644 index 00000000..d586a797 --- /dev/null +++ b/backend/src/gst/rtp_hdrext.rs @@ -0,0 +1,271 @@ +//! Disable RTP header-extension aggregation on every depayloader in a pipeline. +//! +//! Works around an unfixed abort in `GstRTPBaseDepayload` +//! ([gstreamer#5057](https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/5057)): +//! +//! ```text +//! gstrtpbasedepayload.c:942:gst_rtp_base_depayload_handle_buffer: +//! 'gst_buffer_list_length (priv->hdrext_buffers) == 0' should be TRUE +//! ``` +//! +//! Since 1.24 the base class caches the RTP header of every packet feeding the +//! output buffer it is assembling, clearing the cache only when the subclass +//! pushes or flushes. `gst_rtp_base_depayload_delayed()` means "this packet's +//! header belongs to the next output buffer", and the base class asserts the +//! cache is empty when that happens. `rtph264depay` breaks the invariant: an +//! interrupted fragmentation unit calls `delayed()` and then +//! `finish_fragmentation_unit()`, which in access-unit mode can absorb the +//! truncated NAL without producing an output buffer. Nothing is pushed, the +//! cache is still populated, and the process aborts — taking every unrelated +//! flow on the server with it, since `g_assert_true` is not defusable. +//! +//! Turning aggregation off restores the pre-1.24 behaviour: header extensions +//! are read from the current packet instead of accumulated. Strom reads no +//! header-extension metadata, and the extensions that matter for transport +//! (transport-cc, abs-send-time, mid) are consumed by `webrtcbin` well upstream +//! of any depayloader, so nothing is lost. +//! +//! There is no GObject property for this — the C setter is the only switch, and +//! it is `Since: 1.24`. Binding it normally would raise the workspace build +//! floor from 1.22, which would break the default install on Debian 12 and the +//! default ARM64 cross-compile target (Raspberry Pi OS 12), both of which ship +//! GStreamer 1.22. Resolving the symbol at runtime keeps one binary working on +//! both: below 1.24 the lookup fails and we do nothing, which is correct +//! because aggregation — and therefore the bug — does not exist there. + +use gstreamer as gst; +use gstreamer::glib::translate::ToGlibPtr; +use gstreamer::prelude::*; +use std::ffi::c_void; +use std::sync::OnceLock; +use tracing::{debug, info}; + +const SETTER: &str = "gst_rtp_base_depayload_set_aggregate_hdrext_enabled"; +const GETTER: &str = "gst_rtp_base_depayload_is_aggregate_hdrext_enabled"; + +/// `void (GstRTPBaseDepayload *depayload, gboolean enable)` +type SetAggregateFn = unsafe extern "C" fn(*mut gst::ffi::GstElement, i32); +/// `gboolean (GstRTPBaseDepayload *depayload)` +type IsAggregateFn = unsafe extern "C" fn(*mut gst::ffi::GstElement) -> i32; + +/// Resolve a symbol from the already-loaded GStreamer RTP library. +/// +/// `gstreamer-rtp-sys` is a build dependency of the binary, so `libgstrtp-1.0` +/// is loaded and its symbols are in scope before this runs. +#[cfg(unix)] +fn lookup(name: &str) -> Option<*mut c_void> { + let cname = std::ffi::CString::new(name).ok()?; + // SAFETY: `cname` is a valid NUL-terminated string for the duration of the + // call. `dlsym` with RTLD_DEFAULT only reads the global symbol table and + // returns a plain address or NULL; it does not take ownership of anything. + let addr = unsafe { libc::dlsym(libc::RTLD_DEFAULT, cname.as_ptr()) }; + (!addr.is_null()).then_some(addr) +} + +#[cfg(windows)] +fn lookup(name: &str) -> Option<*mut c_void> { + const FROM_ADDRESS: u32 = 0x0000_0004; + const UNCHANGED_REFCOUNT: u32 = 0x0000_0002; + + extern "system" { + fn GetModuleHandleExA(flags: u32, name: *const i8, module: *mut *mut c_void) -> i32; + fn GetProcAddress(module: *mut c_void, name: *const i8) -> *mut c_void; + } + + let cname = std::ffi::CString::new(name).ok()?; + // Anchor on a symbol we already link from the same DLL rather than + // hardcoding its filename, which varies between GStreamer builds. + let anchor = gstreamer_rtp::ffi::gst_rtp_base_depayload_get_type as *const c_void; + let mut module: *mut c_void = std::ptr::null_mut(); + + // SAFETY: `anchor` is the address of a function in the loaded RTP DLL, and + // `module` is a valid out-pointer. UNCHANGED_REFCOUNT means we take no + // reference, so there is nothing to release. `GetProcAddress` reads the + // module's export table and returns a plain address or NULL. + unsafe { + if GetModuleHandleExA( + FROM_ADDRESS | UNCHANGED_REFCOUNT, + anchor as *const i8, + &mut module, + ) == 0 + { + return None; + } + let addr = GetProcAddress(module, cname.as_ptr()); + (!addr.is_null()).then_some(addr) + } +} + +#[cfg(not(any(unix, windows)))] +fn lookup(_name: &str) -> Option<*mut c_void> { + None +} + +fn setter() -> Option { + static SYM: OnceLock> = OnceLock::new(); + let addr = SYM.get_or_init(|| { + let found = lookup(SETTER).map(|p| p as usize); + match found { + Some(_) => debug!("{SETTER} resolved; hdrext aggregation will be disabled"), + None => info!( + "{SETTER} not found (GStreamer < 1.24) - header extension aggregation \ + does not exist on this version, nothing to disable" + ), + } + found + }); + let addr = (*addr)?; + // SAFETY: the address came from resolving `SETTER` in the loaded RTP + // library, whose signature has been stable since 1.24. + Some(unsafe { std::mem::transmute::(addr) }) +} + +fn getter() -> Option { + static SYM: OnceLock> = OnceLock::new(); + let addr = (*SYM.get_or_init(|| lookup(GETTER).map(|p| p as usize)))?; + // SAFETY: as above, for the matching getter. + Some(unsafe { std::mem::transmute::(addr) }) +} + +/// Turn aggregation off on one depayloader. No-op below GStreamer 1.24. +fn disable_on(depay: &gstreamer_rtp::RTPBaseDepayload) { + let Some(set) = setter() else { return }; + let ptr: *mut gst::ffi::GstElement = depay.upcast_ref::().to_glib_none().0; + // SAFETY: `ptr` is a live GstRTPBaseDepayload borrowed for this call — the + // downcast to `RTPBaseDepayload` guarantees the type, and `depay` keeps it + // alive. The setter only writes a boolean field and clears an internal + // buffer list. + unsafe { set(ptr, 0) }; +} + +/// Read aggregation state back. `None` below GStreamer 1.24. +pub fn is_enabled(depay: &gstreamer_rtp::RTPBaseDepayload) -> Option { + let get = getter()?; + let ptr: *mut gst::ffi::GstElement = depay.upcast_ref::().to_glib_none().0; + // SAFETY: as in `disable_on`; the getter only reads a boolean field. + Some(unsafe { get(ptr) } != 0) +} + +/// Whether this GStreamer build has the aggregation switch at all. +pub fn is_supported() -> bool { + setter().is_some() +} + +/// Disable header-extension aggregation on every depayloader the pipeline ever +/// contains, including ones `decodebin` autoplugs and ones a user places by +/// hand in a flow. +/// +/// Install before the pipeline leaves NULL so no depayloader is missed. +pub fn install(pipeline: &gst::Pipeline) { + if !is_supported() { + return; + } + + let bin = pipeline.upcast_ref::(); + + // Elements already present (a depayloader placed directly in a flow is + // built before start()); `deep-element-added` only fires for later ones. + for element in bin.iterate_recurse().into_iter().flatten() { + if let Some(depay) = element.downcast_ref::() { + disable_on(depay); + } + } + + // NOTE: this closure captures nothing at all - no pipeline, no element, no + // map - so it cannot create a reference cycle that keeps the pipeline + // alive. Keep it that way. + bin.connect("deep-element-added", false, move |args| { + let added: gst::Element = args.get(2)?.get().ok()?; + if let Some(depay) = added.downcast_ref::() { + debug!( + "Disabling RTP header extension aggregation on {} (gstreamer#5057)", + added.name() + ); + disable_on(depay); + } + None + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn init() { + let _ = gst::init(); + } + + fn depay() -> gstreamer_rtp::RTPBaseDepayload { + gst::ElementFactory::make("rtph264depay") + .build() + .expect("rtph264depay is in gstreamer1.0-plugins-good, installed in CI") + .downcast::() + .expect("rtph264depay derives from GstRTPBaseDepayload") + } + + /// The whole workaround hinges on resolving the symbol at runtime. If this + /// fails on a >= 1.24 build, `install()` silently does nothing. + #[test] + fn symbol_resolves_on_modern_gstreamer() { + init(); + let (major, minor, ..) = gst::version(); + if (major, minor) >= (1, 24) { + assert!( + is_supported(), + "GStreamer {major}.{minor} has the aggregation API but the runtime \ + lookup failed - the workaround would silently no-op" + ); + } + } + + /// `rtph264depay` opts into aggregation in its _init, so this documents the + /// unpatched default and proves the getter reads real state. + #[test] + fn rtph264depay_enables_aggregation_by_default() { + init(); + if !is_supported() { + return; + } + assert_eq!(is_enabled(&depay()), Some(true)); + } + + /// Guard: a depayloader already in the pipeline when `install()` runs. + /// Fails if the `iterate_recurse` sweep is removed. + #[test] + fn install_disables_aggregation_on_existing_depayloader() { + init(); + if !is_supported() { + return; + } + let pipeline = gst::Pipeline::new(); + let d = depay(); + pipeline.add(&d).unwrap(); + assert_eq!(is_enabled(&d), Some(true), "precondition"); + + install(&pipeline); + + assert_eq!(is_enabled(&d), Some(false)); + } + + /// Guard: a depayloader added to a nested bin *after* `install()` - the + /// shape `decodebin` produces when it autoplugs one. Fails if the + /// `deep-element-added` handler is removed. + #[test] + fn install_disables_aggregation_on_later_nested_depayloader() { + init(); + if !is_supported() { + return; + } + let pipeline = gst::Pipeline::new(); + let inner = gst::Bin::builder().name("inner").build(); + pipeline.add(&inner).unwrap(); + + install(&pipeline); + + // Added after install, one level down - only deep-element-added sees it. + let d = depay(); + inner.add(&d).unwrap(); + + assert_eq!(is_enabled(&d), Some(false)); + } +} diff --git a/backend/tests/rtp_hdrext_aggregation_test.rs b/backend/tests/rtp_hdrext_aggregation_test.rs new file mode 100644 index 00000000..a3a78b87 --- /dev/null +++ b/backend/tests/rtp_hdrext_aggregation_test.rs @@ -0,0 +1,116 @@ +//! Regression test: every RTP depayloader in a started pipeline must have +//! header-extension aggregation disabled. +//! +//! Leaving it on lets an interrupted H264 fragmentation unit trip a `g_assert` +//! in `GstRTPBaseDepayload` (gstreamer#5057, `gstrtpbasedepayload.c:942`) that +//! aborts the whole process, killing every unrelated flow on the server. The +//! switch is `Since: 1.24` and has no GObject property, so `gst::rtp_hdrext` +//! resolves the C setter at runtime and installs a `deep-element-added` handler +//! from `PipelineManager::start()`. +//! +//! The unit tests in `gst::rtp_hdrext` cover `install()` directly. This one +//! covers the wiring: it fails if the `rtp_hdrext::install()` call is removed +//! from `start()`, which the unit tests cannot see. +//! +//! `rtph264depay` ships in `gstreamer1.0-plugins-good`, installed in every +//! Linux CI job, so this runs rather than skipping. + +use gstreamer::prelude::*; +use std::collections::HashMap; +use strom::blocks::BlockRegistry; +use strom::events::EventBroadcaster; +use strom::gst::pipeline::PipelineManager; +use strom::gst::rtp_hdrext; +use strom_types::{Flow, Link}; +use tempfile::NamedTempFile; + +/// `fakesrc → rtph264depay → fakesink`. +/// +/// A user can place `rtph264depay` directly in a flow — block construction +/// accepts any element type name — so this is a real topology, not a contrivance. +/// The pipeline is not expected to run: nothing feeds the depayloader valid RTP. +/// That does not matter, because `install()` runs at the top of `start()`, +/// before any state change. +fn build_depayloader_flow(name: &str) -> Flow { + let mut flow = Flow::new(name); + + for (id, element_type, x) in [ + ("src", "fakesrc", 100.0), + ("depay", "rtph264depay", 250.0), + ("sink", "fakesink", 400.0), + ] { + flow.elements.push(strom_types::Element { + id: id.to_string(), + element_type: element_type.to_string(), + properties: HashMap::new(), + position: [x, 200.0].into(), + pad_properties: HashMap::new(), + }); + } + + flow.links.push(Link { + from: "src:src".to_string(), + to: "depay:sink".to_string(), + }); + flow.links.push(Link { + from: "depay:src".to_string(), + to: "sink:sink".to_string(), + }); + + flow +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_start_disables_hdrext_aggregation_on_depayloaders() { + gstreamer::init().unwrap(); + + if !rtp_hdrext::is_supported() { + // GStreamer < 1.24: aggregation does not exist, so neither does the bug. + return; + } + + let temp_file = NamedTempFile::new().unwrap(); + let registry = BlockRegistry::new(temp_file.path()); + let events = EventBroadcaster::new(10); + + let flow = build_depayloader_flow("hdrext_aggregation_test"); + + let mut manager = PipelineManager::new( + &flow, + events, + ®istry, + vec![], + "all".to_string(), + None, + std::env::temp_dir(), + std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())), + ) + .expect("Failed to create PipelineManager"); + + // start() may fail — nothing feeds the depayloader real RTP — but + // install() runs before any state change, which is what we are asserting. + let _ = manager.start(); + + let mut checked = 0; + for element in manager.pipeline().iterate_recurse().into_iter().flatten() { + if let Ok(depay) = element.downcast::() { + assert_eq!( + rtp_hdrext::is_enabled(&depay), + Some(false), + "depayloader {} still has RTP header extension aggregation enabled \ + after start() — an interrupted H264 fragmentation unit will abort \ + the whole process (gstreamer#5057)", + depay.name() + ); + checked += 1; + } + } + + assert_eq!( + checked, 1, + "expected exactly one depayloader in the pipeline, found {checked} — \ + the test topology no longer exercises what it claims to" + ); + + let _ = manager.stop(); +} From 55e91efad9117373071f03f93567df8f8ca5a046 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Fri, 28 Aug 2026 07:21:57 -0700 Subject: [PATCH 2/2] fix(rtp): install the hdrext workaround on every pipeline, not just the main one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The abort in GstRTPBaseDepayload (gstreamer#5057) kills the process, so it does not matter which pipeline hosts the depayloader that trips it. Three pipelines can host one, and only the main pipeline was covered: - Media Player's internal pipelines autoplug a depayloader from an rtsp:// URI through uridecodebin/urisourcebin. - whipserversrc autoplugs depayloaders inside its own bin. install() is a no-op below GStreamer 1.24 and only clears a boolean, and all three call sites run while the pipeline is still NULL, so no depayloader is missed. The two new tests call the real Media Player constructors and then add a depayloader to a nested bin — the path deep-element-added sees, since CI cannot serve RTSP. Both fail with the install() calls removed (verified). There is no equivalent test for the WHIP path: build_whipserversrc binds a port and needs a full endpoint config. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KcQpknvxTmHSLzsJPaeq61 --- .../src/blocks/builtin/mediaplayer/bridge.rs | 99 +++++++++++++++++++ backend/src/blocks/builtin/whip.rs | 6 ++ 2 files changed, 105 insertions(+) diff --git a/backend/src/blocks/builtin/mediaplayer/bridge.rs b/backend/src/blocks/builtin/mediaplayer/bridge.rs index 101149cb..3ca109db 100644 --- a/backend/src/blocks/builtin/mediaplayer/bridge.rs +++ b/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 { + 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::() + .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); + } +} diff --git a/backend/src/blocks/builtin/whip.rs b/backend/src/blocks/builtin/whip.rs index 592eee6d..135baec7 100644 --- a/backend/src/blocks/builtin/whip.rs +++ b/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)