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/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) 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(); +}