From 5e1d87c70914387d7e89a62e116baec09183ba2f Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Wed, 9 Sep 2026 19:11:35 -0700 Subject: [PATCH 1/2] fix(whip): stop a session's second stream stalling below PLAYING MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A WHIP session pipeline is already PLAYING when whipserversrc adds its pads, and each pad gets a `tee -> fakesink + appsink` branch synced into it. The appsink ran with `GstBaseSink`'s default `async=true`, so it answered that state change with ASYNC and waited for a preroll. The pads arrive one after the other within a millisecond, so the second branch lands while the first is still waiting, and the second is left below PLAYING: it takes a single buffer and then blocks its streaming thread for good. That seat's audio or video then never reaches the slot's appsrc, so its decodebin exposes no pad, and the flow's recorder and mixer wait on a track that will never arrive. Nothing in the log says so — `sync_state_with_parent` reports the state change as accepted either way. Turn `async` off, as the fakesink beside it already does. Measured on a five-seat meeting rig, 18 fresh-server trials per arm: 7/18 seats failed to start before, 0/18 after. The branch construction moves into `attach_session_branch` so the regression test builds the real thing rather than a copy of it. Co-Authored-By: Claude Opus 5 (1M context) --- backend/src/blocks/builtin/whip.rs | 130 +++++++++------ backend/tests/whip_session_branch_test.rs | 183 ++++++++++++++++++++++ 2 files changed, 266 insertions(+), 47 deletions(-) create mode 100644 backend/tests/whip_session_branch_test.rs diff --git a/backend/src/blocks/builtin/whip.rs b/backend/src/blocks/builtin/whip.rs index 6d218b258..71dab444f 100644 --- a/backend/src/blocks/builtin/whip.rs +++ b/backend/src/blocks/builtin/whip.rs @@ -599,6 +599,83 @@ fn wait_for_inactivity( } } +/// Drain one `whipserversrc` pad into the session pipeline: `pad → tee → +/// fakesink + appsink`, with every element brought up to the pipeline's state. +/// +/// Returns the appsink the caller installs its bridge callbacks on, or `None` +/// if the branch could not be built — every failure is logged here. +/// +/// Both sinks run with `async` off. The pipeline they join is already PLAYING, +/// and a sink that still wants a preroll answers the state change with ASYNC. +/// whipserversrc adds its audio and video pads one after the other within a +/// millisecond, so the second branch lands while the first is still waiting, +/// and the second is then left below PLAYING: it takes a single buffer and +/// blocks its streaming thread for good. From the outside that is a publisher +/// whose audio or video never starts, with nothing in the log to say so — +/// `sync_state_with_parent` reports the state change as accepted either way. +pub fn attach_session_branch( + session_pipeline: &gst::Pipeline, + pad: &gst::Pad, + appsink_name: &str, +) -> Option { + let tee = match gst::ElementFactory::make("tee") + .property("allow-not-linked", true) + .build() + { + Ok(t) => t, + Err(e) => { + error!("WHIP Input: Failed to create tee in pad-added: {}", e); + return None; + } + }; + let fakesink = match gst::ElementFactory::make("fakesink") + .property("sync", false) + .property("async", false) + .build() + { + Ok(f) => f, + Err(e) => { + error!("WHIP Input: Failed to create fakesink in pad-added: {}", e); + return None; + } + }; + let appsink = gst_app::AppSink::builder() + .name(appsink_name) + .sync(false) + .async_(false) + .build(); + + if let Err(e) = session_pipeline.add_many([&tee, &fakesink, appsink.upcast_ref()]) { + error!( + "WHIP Input: Failed to add elements to session pipeline: {}", + e + ); + return None; + } + if let Err(e) = pad.link(&tee.static_pad("sink").expect("tee has no sink pad")) { + error!("WHIP Input: Failed to link pad to tee: {:?}", e); + return None; + } + if let (Some(tee_src1), Some(tee_src2)) = ( + tee.request_pad_simple("src_%u"), + tee.request_pad_simple("src_%u"), + ) { + let _ = tee_src1.link( + &fakesink + .static_pad("sink") + .expect("fakesink has no sink pad"), + ); + let _ = tee_src2.link(&appsink.static_pad("sink").expect("appsink has no sink pad")); + } else { + error!("WHIP Input: Failed to request tee src pads"); + return None; + } + let _ = tee.sync_state_with_parent(); + let _ = fakesink.sync_state_with_parent(); + let _ = appsink.sync_state_with_parent(); + Some(appsink) +} + /// Create a new whipserversrc element for a single WHIP client session. /// /// Each session runs in its own isolated GStreamer pipeline to avoid @@ -878,54 +955,13 @@ pub fn create_whipserversrc_for_session( return; }; - // Session pipeline: pad → tee → fakesink (drain) + appsink (bridge) - let tee = match gst::ElementFactory::make("tee") - .property("allow-not-linked", true) - .build() - { - Ok(t) => t, - Err(e) => { - error!("WHIP Input: Failed to create tee in pad-added: {}", e); - return; - } - }; - let fakesink = match gst::ElementFactory::make("fakesink") - .property("sync", false) - .property("async", false) - .build() - { - Ok(f) => f, - Err(e) => { - error!("WHIP Input: Failed to create fakesink in pad-added: {}", e); - return; - } - }; - let appsink = gst_app::AppSink::builder() - .name(format!("{}:{}_appsink_{}", prefix, pad_name, stream_num)) - .sync(false) - .build(); - - if let Err(e) = session_pipeline.add_many([&tee, &fakesink, appsink.upcast_ref()]) { - error!("WHIP Input: Failed to add elements to session pipeline: {}", e); - return; - } - if let Err(e) = pad.link(&tee.static_pad("sink").expect("tee has no sink pad")) { - error!("WHIP Input: Failed to link pad to tee: {:?}", e); + let Some(appsink) = attach_session_branch( + &session_pipeline, + pad, + &format!("{}:{}_appsink_{}", prefix, pad_name, stream_num), + ) else { return; - } - if let (Some(tee_src1), Some(tee_src2)) = ( - tee.request_pad_simple("src_%u"), - tee.request_pad_simple("src_%u"), - ) { - let _ = tee_src1.link(&fakesink.static_pad("sink").expect("fakesink has no sink pad")); - let _ = tee_src2.link(&appsink.static_pad("sink").expect("appsink has no sink pad")); - } else { - error!("WHIP Input: Failed to request tee src pads"); - return; - } - let _ = tee.sync_state_with_parent(); - let _ = fakesink.sync_state_with_parent(); - let _ = appsink.sync_state_with_parent(); + }; // Determine which slot appsrc to feed based on pad type let target_appsrc: Option = diff --git a/backend/tests/whip_session_branch_test.rs b/backend/tests/whip_session_branch_test.rs new file mode 100644 index 000000000..365a7a888 --- /dev/null +++ b/backend/tests/whip_session_branch_test.rs @@ -0,0 +1,183 @@ +//! A WHIP session's audio and video pads appear one after the other within a +//! millisecond, each already carrying media, and each gets a +//! `tee -> fakesink + appsink` branch added to a session pipeline that is +//! already PLAYING. +//! +//! A sink that still wants a preroll answers that state change with ASYNC. Two +//! of those in quick succession on one pipeline leave the second branch below +//! PLAYING, where it takes a single buffer and then blocks its streaming thread +//! for good. From the outside that is a publisher whose audio or video never +//! starts, with nothing in the log to say so — and the seat is useless either +//! way, because the flow's recorder and mixer are waiting on both streams. +//! +//! `attach_session_branch` builds both sinks with `async` off, so there is no +//! ASYNC cycle for the second branch to land in. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use strom::blocks::builtin::whip::attach_session_branch; + +use gstreamer as gst; +use gstreamer::prelude::*; +use gstreamer_app as gst_app; + +/// Elements this test needs beyond core GStreamer. Missing on a bare CI image. +const REQUIRED: &[&str] = &["tee", "fakesink", "appsink", "videotestsrc", "audiotestsrc"]; + +/// One round is enough to strand a branch here — 20 runs of each arm, all 20 +/// failing with the fix reverted and all 20 passing with it. A few rounds are +/// cheap insurance in case a faster or slower host makes it probabilistic +/// again, which is what it is on the rig. +const ROUNDS: usize = 3; + +/// Skipping on a missing element passes green and guards nothing, so CI sets +/// `STROM_REQUIRE_GST_PLUGINS=1` to turn a skip into a failure. +fn plugins_available() -> bool { + let missing: Vec<&str> = REQUIRED + .iter() + .copied() + .filter(|e| gst::ElementFactory::find(e).is_none()) + .collect(); + if missing.is_empty() { + return true; + } + assert!( + strom_types::env::var_opt("STROM_REQUIRE_GST_PLUGINS").is_none(), + "STROM_REQUIRE_GST_PLUGINS is set but these elements are missing: {}", + missing.join(", ") + ); + false +} + +/// A running source standing in for one of whipserversrc's pads, which carry +/// media by the time they appear. +/// +/// The pad handed out is a `tee`'s, not the source's own: `allow-not-linked` +/// lets it drop data until the branch arrives, where a source pushing into its +/// own unlinked pad would stop with a flow error first. +struct LiveSource { + pad: gst::Pad, +} + +impl LiveSource { + fn new(pipeline: &gst::Pipeline, factory: &str) -> Self { + let element = gst::ElementFactory::make(factory) + .property("is-live", true) + .build() + .unwrap_or_else(|_| panic!("{factory} available")); + let tee = gst::ElementFactory::make("tee") + .property("allow-not-linked", true) + .build() + .expect("tee"); + pipeline.add_many([&element, &tee]).expect("add source"); + element.link(&tee).expect("source -> tee"); + let pad = tee.request_pad_simple("src_%u").expect("tee src pad"); + Self { pad } + } +} + +/// Count what an appsink actually receives. Pulling is what the real bridge +/// callback does; without it the sink's queue would fill and hide a stall. +fn count_samples(appsink: &gst_app::AppSink) -> Arc { + let seen = Arc::new(AtomicUsize::new(0)); + let seen_cb = Arc::clone(&seen); + appsink.set_callbacks( + gst_app::AppSinkCallbacks::builder() + .new_sample(move |sink| { + let _ = sink.pull_sample().map_err(|_| gst::FlowError::Eos)?; + seen_cb.fetch_add(1, Ordering::Relaxed); + Ok(gst::FlowSuccess::Ok) + }) + .build(), + ); + seen +} + +/// One session: two pads carrying media, each given a branch on a pipeline that +/// is already PLAYING. Reports each branch's sample count and the state its +/// appsink ended in, so a failure can name both. +fn run_round() -> ((usize, gst::State), (usize, gst::State)) { + let pipeline = gst::Pipeline::new(); + let video = LiveSource::new(&pipeline, "videotestsrc"); + let audio = LiveSource::new(&pipeline, "audiotestsrc"); + + pipeline + .set_state(gst::State::Playing) + .expect("pipeline accepts PLAYING"); + let (_, current, _) = pipeline.state(gst::ClockTime::from_seconds(10)); + assert_eq!( + current, + gst::State::Playing, + "the session pipeline must be PLAYING before its branches are attached" + ); + + // One after the other on one thread, as whipserversrc's pad-added fires. + let sinks: Vec = [(&video.pad, "video_0"), (&audio.pad, "audio_0")] + .into_iter() + .map(|(pad, name)| { + attach_session_branch(&pipeline, pad, name) + .unwrap_or_else(|| panic!("{name} branch attaches")) + }) + .collect(); + + let counts: Vec> = sinks.iter().map(count_samples).collect(); + + // A stranded branch takes one buffer and stops, so wait long enough that + // "one" cannot be confused with "still starting". + let deadline = Instant::now() + Duration::from_secs(3); + while Instant::now() < deadline { + if counts.iter().all(|c| c.load(Ordering::Relaxed) > 5) { + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + + let out = ( + ( + counts[0].load(Ordering::Relaxed), + sinks[0].upcast_ref::().current_state(), + ), + ( + counts[1].load(Ordering::Relaxed), + sinks[1].upcast_ref::().current_state(), + ), + ); + + pipeline + .set_state(gst::State::Null) + .expect("pipeline to NULL"); + out +} + +/// The regression. Both of a session's branches must keep carrying data after +/// being attached, back to back, to a pipeline that is already running. +/// +/// Revert the fix and the branch attached second is left in READY having taken +/// a single buffer, while the first runs normally. Which stream that is on the +/// rig depends on the order whipserversrc exposes its pads, and both orders +/// occur. +#[test] +fn both_session_branches_keep_running_when_attached_in_succession() { + gst::init().expect("gstreamer init"); + if !plugins_available() { + eprintln!("skipping: required GStreamer elements missing"); + return; + } + + for round in 0..ROUNDS { + let ((video_samples, video_state), (audio_samples, audio_state)) = run_round(); + assert!( + video_samples > 5 && audio_samples > 5, + "round {round}: a branch stopped carrying data — \ + video {video_samples} samples in {video_state:?}, \ + audio {audio_samples} samples in {audio_state:?}" + ); + assert_eq!( + (video_state, audio_state), + (gst::State::Playing, gst::State::Playing), + "round {round}: a branch was left below PLAYING" + ); + } +} From 3973b08748e24b0ba108735f966e7e7481ad69a1 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Thu, 10 Sep 2026 09:05:41 -0700 Subject: [PATCH 2/2] fix(thumbnail): turn off async on the thumbnail tap appsink The thumbnail branch is attached to a flow pipeline that is already running, and its appsink ran with GstBaseSink's default async=true, so it answered that state change with ASYNC and waited for a preroll. The activation guard admits the pipeline in PAUSED with PLAYING pending, which is that wait, and a branch attached inside one is left below PLAYING: it takes a single buffer and blocks its streaming thread. Activation returns without waiting for a preroll, and a branch is created per block from an HTTP request, so a client asking for several blocks' thumbnails serializes on one mutex but not on the async cycles behind it. Same defect and same fix as the WHIP session branch. Co-Authored-By: Claude Opus 5 (1M context) --- backend/src/gst/thumbnail_tap.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/backend/src/gst/thumbnail_tap.rs b/backend/src/gst/thumbnail_tap.rs index fff720976..43457517f 100644 --- a/backend/src/gst/thumbnail_tap.rs +++ b/backend/src/gst/thumbnail_tap.rs @@ -285,11 +285,18 @@ impl ThumbnailTap { .build() .map_err(|e| ThumbnailError::FrameMapping(format!("capsfilter: {}", e)))?; + // `async` off: the branch joins a pipeline that is already running, and + // a sink still wanting a preroll answers that state change with ASYNC, + // taking the pipeline back to PAUSED until a frame arrives. Activation + // is allowed during that wait, and a branch attached inside one never + // completes its own state change: it takes a single buffer, then blocks + // its streaming thread. let appsink = gst_app::AppSink::builder() .name(format!("{}_thumb_sink", prefix)) .max_buffers(1) .drop(true) .sync(false) + .async_(false) .build(); // Set up appsink callback — pad probe on queue src limits fps,