From a32d27f163c84ec2544868fd8260fbf450bd38b1 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Sun, 13 Sep 2026 21:49:56 -0700 Subject: [PATCH 1/7] fix(mpegtssrt_input): re-anchor tsdemux stamped ahead by a stale head With ignore-pcr, tsdemux maps PTS to output time from the first PES it parses and never takes a new reference. An SRT caller that encoded while it waited for the listener sends a few frames from the start of its encode, then live frames. The stale frame becomes the reference, so live audio and video are stamped as far into the future as the caller waited, and every downstream sink and mixer holds the source back by that much for the whole connection. TsDemuxAnchor compares each demuxed buffer's running time with the latest srtsrc timestamp (arrival less SRT transit). When the smallest lead over a 500 ms window exceeds 200 ms, it moves all of the demuxer's source pads back by the window's mean lead with one shared pad offset. The probes use buffer timestamps and atomics only. PCR mode is left alone. Co-Authored-By: Claude Opus 5 (1M context) --- backend/src/blocks/builtin/mod.rs | 1 + backend/src/blocks/builtin/mpegtssrt_input.rs | 21 ++ backend/src/blocks/builtin/tsdemux_anchor.rs | 250 ++++++++++++++ .../tests/mpegtssrt_input_stale_head_test.rs | 315 ++++++++++++++++++ 4 files changed, 587 insertions(+) create mode 100644 backend/src/blocks/builtin/tsdemux_anchor.rs create mode 100644 backend/tests/mpegtssrt_input_stale_head_test.rs diff --git a/backend/src/blocks/builtin/mod.rs b/backend/src/blocks/builtin/mod.rs index 9c8b964c4..55a21512b 100644 --- a/backend/src/blocks/builtin/mod.rs +++ b/backend/src/blocks/builtin/mod.rs @@ -32,6 +32,7 @@ pub mod spectrum; pub mod tams_output; pub mod thumbnail; pub mod time_offset; +pub mod tsdemux_anchor; pub mod videoenc; pub mod videoformat; pub mod vision_mixer; diff --git a/backend/src/blocks/builtin/mpegtssrt_input.rs b/backend/src/blocks/builtin/mpegtssrt_input.rs index 134391e6b..dad205b3a 100644 --- a/backend/src/blocks/builtin/mpegtssrt_input.rs +++ b/backend/src/blocks/builtin/mpegtssrt_input.rs @@ -18,6 +18,7 @@ //! Both `decodebin` and `tsdemux` have dynamic pads — uses `connect_pad_added` //! to link to identity elements based on caps (video/ or audio/). +use super::tsdemux_anchor::TsDemuxAnchor; use crate::blocks::{BlockBuildContext, BlockBuildError, BlockBuildResult, BlockBuilder}; use gstreamer as gst; use gstreamer::prelude::*; @@ -218,6 +219,19 @@ impl BlockBuilder for MpegTsSrtInputBuilder { srt_uri, latency, tsdemux_latency, ignore_pcr, keep_listening, auto_reconnect, wait_for_connection ); + // With ignore-pcr, tsdemux keeps the time reference of the first PES it + // sees; a caller that connects with stale frames at its head is stamped + // into the future for the whole connection. + let anchor = if ignore_pcr { + let anchor = TsDemuxAnchor::new(instance_id); + if let Some(src_pad) = srtsrc.static_pad("src") { + anchor.watch_input(&src_pad); + } + Some(anchor) + } else { + None + }; + // Create demux/decode element // NOTE (2026-03-02): Using decodebin (v2) instead of decodebin3 because decodebin3 // has known issues with MPEG-TS where it only exposes audio and skips video due to @@ -232,6 +246,7 @@ impl BlockBuilder for MpegTsSrtInputBuilder { // Catch tsdemux when decodebin creates it internally, and set its properties. let instance_for_deep = instance_id.to_string(); + let anchor_for_deep = anchor.clone(); element.connect("deep-element-added", false, move |args| { let added: gst::Element = args[2].get().unwrap(); let factory = added.factory(); @@ -241,6 +256,9 @@ impl BlockBuilder for MpegTsSrtInputBuilder { if ignore_pcr && added.has_property("ignore-pcr") { added.set_property("ignore-pcr", true); } + if let Some(anchor) = &anchor_for_deep { + anchor.watch_demuxer(&added); + } info!( "MPEGTSSRT Input {}: Set tsdemux latency={}ms, ignore-pcr={} (inside decodebin)", instance_for_deep, tsdemux_latency, ignore_pcr @@ -261,6 +279,9 @@ impl BlockBuilder for MpegTsSrtInputBuilder { if ignore_pcr && element.has_property("ignore-pcr") { element.set_property("ignore-pcr", true); } + if let Some(anchor) = &anchor { + anchor.watch_demuxer(&element); + } (id, element) }; diff --git a/backend/src/blocks/builtin/tsdemux_anchor.rs b/backend/src/blocks/builtin/tsdemux_anchor.rs new file mode 100644 index 000000000..eec11d05a --- /dev/null +++ b/backend/src/blocks/builtin/tsdemux_anchor.rs @@ -0,0 +1,250 @@ +//! Re-anchors `tsdemux` output that runs ahead of the arrival timestamps. +//! +//! With `ignore-pcr=true`, `tsdemux` maps PTS to output time from a single +//! reference: the upstream timestamp of the buffer carrying the first PES it +//! parses (`mpegts_packetizer_pts_to_ts_internal`, "Take the first base time we +//! see as a reference"). Every later PTS is placed relative to that one and the +//! reference is never taken again. +//! +//! That is only right when the first PES is current. An SRT caller that has +//! been encoding before it could connect (listener not up yet, caller retrying) +//! delivers a few frames from the start of its encode, then jumps to live PTS. +//! The stale frames become the reference, so every live frame is stamped the +//! length of that PTS jump into the future — for the life of the connection. +//! Downstream sinks and aggregators hold each buffer until its running time, +//! so the source is delayed by however long the caller waited to connect. +//! +//! `srtsrc` stamps each buffer with its arrival time less the SRT transit +//! delay, so healthy demuxed output never runs ahead of the latest input by +//! more than the few frames one SRT message carries. This watcher compares the +//! two, and when the smallest lead over a window exceeds [`MAX_LEAD`] it shifts +//! all of the demuxer's source pads back by the window's mean lead with a pad +//! offset. One +//! correction is shared by every pad, so audio and video stay aligned. +//! +//! Timing comes only from buffer timestamps, never the clock: the probes are on +//! the streaming thread for every demuxed frame and every SRT message, so they +//! use atomics only and do no locking or allocation outside the rare +//! correction. + +use gstreamer as gst; +use gstreamer::prelude::*; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; +use std::sync::Arc; +use tracing::info; + +/// Smallest lead over one window above which the output is re-anchored. +/// +/// Healthy lead is a sawtooth: every PES completed within one SRT message is +/// placed from the same reference, while the input time is that of the whole +/// message. A 1316-byte message of 128 kbit/s AAC spans about 100 ms, and lower +/// bitrates span more. +pub const MAX_LEAD: gst::ClockTime = gst::ClockTime::from_mseconds(200); + +/// Input time covered by one measurement window. +pub const WINDOW: gst::ClockTime = gst::ClockTime::from_mseconds(500); + +const NONE: u64 = u64::MAX; + +struct State { + instance_id: String, + /// Running time of the most recent input buffer, or `NONE`. + last_input: AtomicU64, + /// Input running time at which the current window started, or `NONE`. + window_start: AtomicU64, + /// Smallest lead seen in the current window (ns). + window_min_lead: AtomicI64, + /// Sum and count of the leads seen in the current window. + window_lead_sum: AtomicI64, + window_samples: AtomicI64, + /// Offset every source pad of the demuxer should carry (ns, never positive). + correction: AtomicI64, +} + +/// Watches one `tsdemux` and the element feeding it. +#[derive(Clone)] +pub struct TsDemuxAnchor { + state: Arc, +} + +impl TsDemuxAnchor { + pub fn new(instance_id: &str) -> Self { + Self { + state: Arc::new(State { + instance_id: instance_id.to_string(), + last_input: AtomicU64::new(NONE), + window_start: AtomicU64::new(NONE), + window_min_lead: AtomicI64::new(i64::MAX), + window_lead_sum: AtomicI64::new(0), + window_samples: AtomicI64::new(0), + correction: AtomicI64::new(0), + }), + } + } + + /// Record the timestamps of buffers entering the demuxer. + /// + /// `pad` must carry a TIME segment starting at zero with timestamps that are + /// already running times, as `srtsrc` produces. + pub fn watch_input(&self, pad: &gst::Pad) { + let state = self.state.clone(); + pad.add_probe(gst::PadProbeType::BUFFER, move |_, info| { + if let Some(pts) = info.buffer().and_then(|b| b.pts()) { + state.last_input.store(pts.nseconds(), Ordering::Relaxed); + } + gst::PadProbeReturn::Ok + }); + } + + /// Watch every source pad the demuxer adds. + /// + /// The handler captures only the shared state, never the element. + pub fn watch_demuxer(&self, tsdemux: &gst::Element) { + let anchor = self.clone(); + tsdemux.connect_pad_added(move |_, pad| { + if pad.direction() == gst::PadDirection::Src { + anchor.watch_output(pad); + } + }); + } + + fn watch_output(&self, pad: &gst::Pad) { + let state = self.state.clone(); + let segment = Arc::new(PadSegment::default()); + pad.add_probe( + gst::PadProbeType::BUFFER | gst::PadProbeType::EVENT_DOWNSTREAM, + move |pad, info| { + match &info.data { + Some(gst::PadProbeData::Buffer(buffer)) => { + if let Some(ts) = buffer.dts_or_pts() { + on_output_buffer(&state, &segment, pad, ts.nseconds()); + } + } + Some(gst::PadProbeData::Event(event)) => { + if let gst::EventView::Segment(e) = event.view() { + segment.record(e.segment()); + } + } + _ => {} + } + gst::PadProbeReturn::Ok + }, + ); + } +} + +/// The last TIME segment seen on one source pad, as plain integers. +struct PadSegment { + /// `start + offset` of the segment (ns), or `NONE` when unusable. + origin: AtomicU64, + base: AtomicU64, + /// Pad offset already folded into that segment by `gst_pad_push_event`. + embedded_offset: AtomicI64, + /// Offset this pad currently carries (ns). + applied_offset: AtomicI64, +} + +impl Default for PadSegment { + fn default() -> Self { + Self { + origin: AtomicU64::new(NONE), + base: AtomicU64::new(0), + embedded_offset: AtomicI64::new(0), + applied_offset: AtomicI64::new(0), + } + } +} + +impl PadSegment { + fn record(&self, segment: &gst::Segment) { + // Pad offsets are applied to a SEGMENT before downstream probes run, so + // this segment already includes whatever offset the pad carries. + let usable = segment + .downcast_ref::() + .filter(|s| s.rate() == 1.0) + .and_then(|s| Some((s.start()?, s.offset()?, s.base()?))); + match usable { + Some((start, offset, base)) => { + self.base.store(base.nseconds(), Ordering::Relaxed); + self.embedded_offset.store( + self.applied_offset.load(Ordering::Relaxed), + Ordering::Relaxed, + ); + self.origin + .store(start.nseconds() + offset.nseconds(), Ordering::Relaxed); + } + None => self.origin.store(NONE, Ordering::Relaxed), + } + } + + /// Running time of `ts` with this pad's current offset, ignoring clipping. + fn running_time(&self, ts: u64) -> Option { + let origin = self.origin.load(Ordering::Relaxed); + if origin == NONE { + return None; + } + let base = self.base.load(Ordering::Relaxed) as i64; + let unshifted = + ts as i64 - origin as i64 + base - self.embedded_offset.load(Ordering::Relaxed); + Some(unshifted + self.applied_offset.load(Ordering::Relaxed)) + } +} + +fn on_output_buffer(state: &State, segment: &PadSegment, pad: &gst::Pad, ts: u64) { + let correction = state.correction.load(Ordering::Relaxed); + if segment.applied_offset.load(Ordering::Relaxed) != correction { + apply(segment, pad, correction); + } + + let input = state.last_input.load(Ordering::Relaxed); + let Some(running_time) = segment.running_time(ts) else { + return; + }; + if input == NONE { + return; + } + let lead = running_time - input as i64; + + let window_start = state.window_start.load(Ordering::Relaxed); + if window_start == NONE || input < window_start { + state.window_start.store(input, Ordering::Relaxed); + state.window_min_lead.store(lead, Ordering::Relaxed); + state.window_lead_sum.store(lead, Ordering::Relaxed); + state.window_samples.store(1, Ordering::Relaxed); + return; + } + let min_lead = lead.min(state.window_min_lead.load(Ordering::Relaxed)); + let lead_sum = state.window_lead_sum.load(Ordering::Relaxed) + lead; + let samples = state.window_samples.load(Ordering::Relaxed) + 1; + if input - window_start < WINDOW.nseconds() { + state.window_min_lead.store(min_lead, Ordering::Relaxed); + state.window_lead_sum.store(lead_sum, Ordering::Relaxed); + state.window_samples.store(samples, Ordering::Relaxed); + return; + } + + state.window_start.store(NONE, Ordering::Relaxed); + if min_lead <= MAX_LEAD.nseconds() as i64 { + return; + } + + // The minimum only detects the jump. Shifting by the mean puts the + // sawtooth where a clean reference leaves it, centred on the arrival time. + let mean_lead = lead_sum / samples; + let correction = correction - mean_lead; + state.correction.store(correction, Ordering::Relaxed); + apply(segment, pad, correction); + info!( + "MPEGTSSRT Input {}: demuxed timestamps ran {} ms ahead of arrival, re-anchoring (total offset {} ms)", + state.instance_id, + mean_lead / 1_000_000, + correction / 1_000_000 + ); +} + +fn apply(segment: &PadSegment, pad: &gst::Pad, offset: i64) { + segment.applied_offset.store(offset, Ordering::Relaxed); + // Takes effect for the buffer being pushed: gst_pad_push_data resends the + // sticky SEGMENT, now offset, after the buffer probes and before the push. + pad.set_offset(offset); +} diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs new file mode 100644 index 000000000..28fd8e8ac --- /dev/null +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -0,0 +1,315 @@ +//! Regression test: an SRT caller whose stream starts with stale frames must not +//! be stamped into the future. +//! +//! A caller that has been encoding while it waited for the listener delivers a +//! few frames from the start of its encode and then jumps to live PTS. With +//! `ignore-pcr`, `tsdemux` takes its time reference from the first frame and +//! places every later frame relative to it, so live audio comes out as far +//! ahead of its arrival as the caller waited — seconds — and every sink and +//! mixer downstream holds it that long. +//! +//! The test replays that over a real SRT connection into the block: a +//! short head from the start of a pre-encoded transport stream, then the same +//! stream from two seconds in, paced in real time. It measures how far each +//! buffer leaving the block runs ahead of the latest SRT input. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use strom::blocks::builtin::mpegtssrt_input::MpegTsSrtInputBuilder; +use strom::blocks::builtin::tsdemux_anchor::{MAX_LEAD, WINDOW}; +use strom::blocks::{BlockBuildContext, BlockBuilder}; +use strom_types::PropertyValue; + +use gstreamer as gst; +use gstreamer::prelude::*; +use gstreamer_app as gst_app; + +const REQUIRED: &[&str] = &[ + "srtsrc", + "srtsink", + "tsdemux", + "mpegtsmux", + "decodebin", + "avenc_aac", + "avdec_aac", + "aacparse", + "audiotestsrc", + "audioconvert", + "audioresample", + "appsrc", + "appsink", + "fakesink", +]; + +/// 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 +} + +fn srt_port() -> u16 { + std::net::UdpSocket::bind("127.0.0.1:0") + .and_then(|s| s.local_addr()) + .map(|a| a.port()) + .expect("no free UDP port for the SRT listener") +} + +/// Where the live part of the replayed stream starts. +const GAP: gst::ClockTime = gst::ClockTime::from_seconds(2); +/// Stale frames sent ahead of the live part. +const HEAD: gst::ClockTime = gst::ClockTime::from_mseconds(150); +/// Live stream replayed after the head. +const LIVE: gst::ClockTime = gst::ClockTime::from_mseconds(2500); + +/// A 128 kbit/s AAC transport stream in 1316-byte chunks, with each chunk's +/// content time, encoded as fast as possible. +fn encode_stream() -> Vec<(gst::ClockTime, gst::Buffer)> { + let seconds = (GAP + LIVE + gst::ClockTime::SECOND).seconds() as i32; + let pipeline = gst::parse::launch(&format!( + "audiotestsrc num-buffers={} samplesperbuffer=4800 wave=ticks \ + ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! avenc_aac bitrate=128000 \ + ! aacparse ! mpegtsmux alignment=7 ! appsink name=out sync=false", + seconds * 10 + )) + .expect("encoder pipeline parses") + .downcast::() + .unwrap(); + let appsink = pipeline + .by_name("out") + .unwrap() + .downcast::() + .unwrap(); + pipeline.set_state(gst::State::Playing).unwrap(); + + let mut chunks = Vec::new(); + let mut last = gst::ClockTime::ZERO; + while let Ok(sample) = appsink.pull_sample() { + let buffer = sample.buffer_owned().unwrap(); + last = buffer.pts().unwrap_or(last); + chunks.push((last, buffer)); + } + pipeline.set_state(gst::State::Null).unwrap(); + assert!( + chunks.last().is_some_and(|(t, _)| *t > GAP + LIVE), + "encoder produced too little stream to replay" + ); + chunks +} + +struct Receiver { + pipeline: gst::Pipeline, + /// (latest SRT input running time, lead of the output buffer) in ns. + leads: Arc>>, +} + +fn build_receiver(decode: bool, port: u16) -> Receiver { + let instance_id = "srt_in"; + let mut props: HashMap = HashMap::new(); + props.insert("decode".to_string(), PropertyValue::Bool(decode)); + props.insert("num_video_tracks".to_string(), PropertyValue::UInt(0)); + props.insert("num_audio_tracks".to_string(), PropertyValue::UInt(1)); + props.insert("latency".to_string(), PropertyValue::Int(20)); + props.insert("keep_listening".to_string(), PropertyValue::Bool(false)); + props.insert( + "srt_uri".to_string(), + PropertyValue::String(format!("srt://127.0.0.1:{}?mode=listener", port)), + ); + + let ctx = BlockBuildContext::new(vec![], "all".to_string()); + let built = MpegTsSrtInputBuilder + .build(instance_id, &props, &ctx) + .expect("mpegts/srt input block builds"); + + let pipeline = gst::Pipeline::new(); + let mut by_id: HashMap = HashMap::new(); + for (id, element) in &built.elements { + pipeline.add(element).expect("add block element"); + by_id.insert(id.clone(), element.clone()); + } + for (from, to) in &built.internal_links { + let src_pad = by_id[&from.element_id] + .static_pad(from.pad_name.as_deref().unwrap()) + .unwrap(); + let sink_pad = by_id[&to.element_id] + .static_pad(to.pad_name.as_deref().unwrap()) + .unwrap(); + src_pad.link(&sink_pad).expect("internal link"); + } + + let sink = gst::ElementFactory::make("fakesink") + .property("sync", false) + .property("async", false) + .build() + .unwrap(); + pipeline.add(&sink).unwrap(); + let output = &by_id[&format!("{}:audio_output_0", instance_id)]; + output.link(&sink).expect("link audio output to fakesink"); + + let last_input = Arc::new(AtomicU64::new(u64::MAX)); + let recorder = last_input.clone(); + by_id[&format!("{}:srtsrc", instance_id)] + .static_pad("src") + .unwrap() + .add_probe(gst::PadProbeType::BUFFER, move |_, info| { + if let Some(pts) = info.buffer().and_then(|b| b.pts()) { + recorder.store(pts.nseconds(), Ordering::Relaxed); + } + gst::PadProbeReturn::Ok + }) + .unwrap(); + + let leads: Arc>> = Arc::default(); + let recorder = leads.clone(); + sink.static_pad("sink") + .unwrap() + .add_probe(gst::PadProbeType::BUFFER, move |pad, info| { + let input = last_input.load(Ordering::Relaxed); + let Some(pts) = info.buffer().and_then(|b| b.pts()) else { + return gst::PadProbeReturn::Ok; + }; + let running_time = pad.sticky_event::(0).and_then(|s| { + s.segment() + .downcast_ref::() + .and_then(|t| t.to_running_time_full(pts)) + }); + if let (Some(rt), true) = (running_time, input != u64::MAX) { + let rt = match rt { + gst::Signed::Positive(t) => t.nseconds() as i64, + gst::Signed::Negative(t) => -(t.nseconds() as i64), + }; + recorder.lock().unwrap().push((input, rt - input as i64)); + } + gst::PadProbeReturn::Ok + }) + .unwrap(); + + Receiver { pipeline, leads } +} + +/// Sends the head, then the live part in real time, over one SRT connection. +/// Returns the sender still connected; see [`stop_srt`] for the teardown order. +fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline { + let pipeline = gst::parse::launch(&format!( + "appsrc name=in is-live=true format=time \ + caps=video/mpegts,systemstream=true,packetsize=188 \ + ! srtsink name=out uri=srt://127.0.0.1:{}?mode=caller latency=20 \ + wait-for-connection=true sync=false", + port + )) + .expect("sender pipeline parses") + .downcast::() + .unwrap(); + let appsrc = pipeline + .by_name("in") + .unwrap() + .downcast::() + .unwrap(); + pipeline.set_state(gst::State::Playing).unwrap(); + + for (_, buffer) in chunks.iter().filter(|(t, _)| *t < HEAD) { + appsrc.push_buffer(buffer.copy()).unwrap(); + } + let start = Instant::now(); + for (t, buffer) in chunks.iter().filter(|(t, _)| *t >= GAP && *t < GAP + LIVE) { + let due = Duration::from_nanos((*t - GAP).nseconds()); + if let Some(wait) = due.checked_sub(start.elapsed()) { + std::thread::sleep(wait); + } + appsrc.push_buffer(buffer.copy()).unwrap(); + } + std::thread::sleep(Duration::from_millis(300)); + pipeline +} + +/// Stops the named SRT element on its own, then the rest of its pipeline. +fn stop_srt(pipeline: &gst::Pipeline, element: &str) { + let _ = pipeline + .by_name(element) + .unwrap() + .set_state(gst::State::Null); + let _ = pipeline.set_state(gst::State::Null); +} + +fn max_live_lead(decode: bool) -> Option<(i64, usize)> { + gst::init().unwrap(); + if !plugins_available() { + eprintln!("skipping: required GStreamer elements are missing"); + return None; + } + + let chunks = encode_stream(); + let port = srt_port(); + let receiver = build_receiver(decode, port); + receiver + .pipeline + .set_state(gst::State::Playing) + .expect("receiver goes to PLAYING"); + + let sender = replay(port, &chunks); + + // Receiver first, while the caller is still connected: an srtsrc whose + // caller has just left can park waiting for the next one and never wake + // for the state change. + stop_srt(&receiver.pipeline, "srt_in:srtsrc"); + stop_srt(&sender, "out"); + + let leads = receiver.leads.lock().unwrap().clone(); + let first_input = leads.first().map(|(i, _)| *i).expect( + "no audio left the block — the SRT stream never decoded, so the test proved nothing", + ); + // The first window closes over the head, whose lead is small; the second + // is all live and triggers the correction. After that every buffer must be + // on time. + let settle = first_input + (2 * WINDOW + gst::ClockTime::from_mseconds(250)).nseconds(); + let settled: Vec = leads + .iter() + .filter(|(input, _)| *input >= settle) + .map(|(_, lead)| *lead) + .collect(); + assert!( + settled.len() > 20, + "only {} buffers after the correction window — too few to judge", + settled.len() + ); + Some((settled.into_iter().max().unwrap(), leads.len())) +} + +fn assert_on_time(decode: bool) { + let Some((max_lead, count)) = max_live_lead(decode) else { + return; + }; + assert!( + max_lead < MAX_LEAD.nseconds() as i64, + "decode={}: live audio left the block {} ms ahead of its SRT arrival \ + ({} buffers). tsdemux took its time reference from the stale head, so \ + every sink downstream holds this source back by the caller's wait.", + decode, + max_lead / 1_000_000, + count + ); +} + +#[test] +fn stale_head_is_not_stamped_ahead_decoded() { + assert_on_time(true); +} + +#[test] +fn stale_head_is_not_stamped_ahead_passthrough() { + assert_on_time(false); +} From 2b0dbd5ee27e111d80f320073b9d22422717a7dd Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Tue, 15 Sep 2026 17:15:54 -0400 Subject: [PATCH 2/7] test(mpegtssrt_input): keep PAT/PMT continuous in the stale-head replay The replay dropped every packet between the head and the live part, which broke the table continuity counters as well as the media. GStreamer 1.24.2, which Ubuntu 24.04 and CI ship, then re-applies the PMT and, with ignore-pcr, mistakes it for a new program (fixed upstream in 1.24.3). tsdemux replaces its source pad, the new pad is unlinked, and srtsrc stops with not-linked after the head, so no buffers reached the correction window. Forward the table packets from the gap and drop only the media. The PTS jump the test guards is unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/mpegtssrt_input_stale_head_test.rs | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index 28fd8e8ac..e6ae73ca4 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -224,6 +224,9 @@ fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline for (_, buffer) in chunks.iter().filter(|(t, _)| *t < HEAD) { appsrc.push_buffer(buffer.copy()).unwrap(); } + for buffer in tables_between(chunks, HEAD, GAP) { + appsrc.push_buffer(buffer).unwrap(); + } let start = Instant::now(); for (t, buffer) in chunks.iter().filter(|(t, _)| *t >= GAP && *t < GAP + LIVE) { let due = Duration::from_nanos((*t - GAP).nseconds()); @@ -236,6 +239,62 @@ fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline pipeline } +const TS_PACKET: usize = 188; + +fn ts_pid(packet: &[u8]) -> u16 { + (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]) +} + +/// Whether a TS packet starts a PES packet, i.e. carries media rather than +/// tables. +fn starts_pes(packet: &[u8]) -> bool { + if packet[1] & 0x40 == 0 { + return false; + } + let payload = match (packet[3] >> 4) & 0x3 { + 1 => 4, + 3 => 5 + usize::from(packet[4]), + _ => return false, + }; + packet.get(payload..payload + 3) == Some(&[0, 0, 1][..]) +} + +/// The table packets (PAT, PMT) of the chunks in `[from, to)`, without media. +/// +/// Keeps the tables continuous across the jump to the live part. GStreamer +/// 1.24.2, which CI runs, re-applies the PMT after a continuity break and, with +/// `ignore-pcr`, mistakes it for a new program (fixed in 1.24.3, "mpegtsbase: +/// Fix Program equality check"). tsdemux then replaces its source pad, the new +/// pad is not linked, and the stream stops before any live audio leaves the +/// block. +fn tables_between( + chunks: &[(gst::ClockTime, gst::Buffer)], + from: gst::ClockTime, + to: gst::ClockTime, +) -> Vec { + let mut media_pids = std::collections::HashSet::new(); + for (_, buffer) in chunks { + let map = buffer.map_readable().unwrap(); + for packet in map.chunks_exact(TS_PACKET).filter(|p| starts_pes(p)) { + media_pids.insert(ts_pid(packet)); + } + } + chunks + .iter() + .filter(|(t, _)| *t >= from && *t < to) + .filter_map(|(_, buffer)| { + let map = buffer.map_readable().unwrap(); + let tables: Vec = map + .chunks_exact(TS_PACKET) + .filter(|p| !media_pids.contains(&ts_pid(p))) + .flatten() + .copied() + .collect(); + (!tables.is_empty()).then(|| gst::Buffer::from_mut_slice(tables)) + }) + .collect() +} + /// Stops the named SRT element on its own, then the rest of its pipeline. fn stop_srt(pipeline: &gst::Pipeline, element: &str) { let _ = pipeline From dede8a3c4b872b5fecf5960834cfe3870b16f0b5 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Wed, 23 Sep 2026 10:39:02 -0700 Subject: [PATCH 3/7] fix(mpegtssrt_input): drop the correction when the SRT caller changes srtsrc keeps listening after a caller leaves and tsdemux takes a fresh time reference for the next one, but the pad offset stayed. A second caller with a clean stream left the block as far behind its arrival as the first caller's stale head had been ahead of it (-2041 ms measured), where every consumer that waits for running time drops it as late. Nothing brought it back, because the offset only ever moved further back. Drop the correction on srtsrc's caller-added and caller-removed. Output that sits behind arrival is still left alone: a sender that stalls and resumes looks the same from here, and giving the offset back would throw its stream forward by that much in one buffer. The replay test gains a second caller in both shapes, resuming and restarting, and the module gains unit tests for what a caller change and a window spent behind arrival each do to the offset. Co-Authored-By: Claude Opus 5 (1M context) --- backend/src/blocks/builtin/mpegtssrt_input.rs | 1 + backend/src/blocks/builtin/tsdemux_anchor.rs | 103 +++++++++++++- .../tests/mpegtssrt_input_stale_head_test.rs | 126 ++++++++++++++++-- 3 files changed, 218 insertions(+), 12 deletions(-) diff --git a/backend/src/blocks/builtin/mpegtssrt_input.rs b/backend/src/blocks/builtin/mpegtssrt_input.rs index dad205b3a..6d79bf169 100644 --- a/backend/src/blocks/builtin/mpegtssrt_input.rs +++ b/backend/src/blocks/builtin/mpegtssrt_input.rs @@ -227,6 +227,7 @@ impl BlockBuilder for MpegTsSrtInputBuilder { if let Some(src_pad) = srtsrc.static_pad("src") { anchor.watch_input(&src_pad); } + anchor.watch_caller(&srtsrc); Some(anchor) } else { None diff --git a/backend/src/blocks/builtin/tsdemux_anchor.rs b/backend/src/blocks/builtin/tsdemux_anchor.rs index eec11d05a..d788f9637 100644 --- a/backend/src/blocks/builtin/tsdemux_anchor.rs +++ b/backend/src/blocks/builtin/tsdemux_anchor.rs @@ -19,8 +19,15 @@ //! more than the few frames one SRT message carries. This watcher compares the //! two, and when the smallest lead over a window exceeds [`MAX_LEAD`] it shifts //! all of the demuxer's source pads back by the window's mean lead with a pad -//! offset. One -//! correction is shared by every pad, so audio and video stay aligned. +//! offset. One correction is shared by every pad, so audio and video stay +//! aligned. +//! +//! The shift belongs to the caller it was measured on: `srtsrc` keeps listening +//! after a caller leaves and `tsdemux` takes a fresh reference for the next one, +//! so the shift is dropped when the caller changes. Output that sits *behind* +//! arrival is left alone, whatever the shift: a sender that stalls and resumes +//! looks the same from here, and giving the shift back would throw its stream +//! forward by that much in one buffer. //! //! Timing comes only from buffer timestamps, never the clock: the probes are on //! the streaming thread for every demuxed frame and every SRT message, so they @@ -61,6 +68,17 @@ struct State { correction: AtomicI64, } +impl State { + /// Drop the correction and the window in progress. + /// + /// The window counters are re-initialised by the first buffer after this, + /// which takes the `window_start == NONE` branch of [`on_output_buffer`]. + fn reset(&self) { + self.correction.store(0, Ordering::Relaxed); + self.window_start.store(NONE, Ordering::Relaxed); + } +} + /// Watches one `tsdemux` and the element feeding it. #[derive(Clone)] pub struct TsDemuxAnchor { @@ -96,6 +114,20 @@ impl TsDemuxAnchor { }); } + /// Drop the correction whenever the SRT caller changes. + /// + /// Listener mode only: `srtsrc` emits these while accepting and dropping a + /// caller, and neither fires in caller mode. + pub fn watch_caller(&self, srtsrc: &gst::Element) { + for signal in ["caller-added", "caller-removed"] { + let state = self.state.clone(); + srtsrc.connect(signal, false, move |_| { + state.reset(); + None + }); + } + } + /// Watch every source pad the demuxer adds. /// /// The handler captures only the shared state, never the element. @@ -248,3 +280,70 @@ fn apply(segment: &PadSegment, pad: &gst::Pad, offset: i64) { // sticky SEGMENT, now offset, after the buffer probes and before the push. pad.set_offset(offset); } + +#[cfg(test)] +mod tests { + use super::*; + + /// A source pad carrying a segment that starts at zero, as `tsdemux` + /// produces, with the timestamps left as running times. + fn pad_and_segment() -> (gst::Pad, PadSegment) { + gst::init().unwrap(); + let pad = gst::Pad::builder(gst::PadDirection::Src) + .name("src") + .build(); + let segment = PadSegment::default(); + let time = gst::FormattedSegment::::new(); + segment.record(time.upcast_ref()); + (pad, segment) + } + + /// One full window of buffers whose running time sits `lead` from arrival. + fn window(state: &State, segment: &PadSegment, pad: &gst::Pad, lead: i64) { + for step in 0..8u64 { + let input = step * 100 * gst::ClockTime::MSECOND.nseconds(); + state.last_input.store(input, Ordering::Relaxed); + // running_time(ts) is ts plus whatever offset the pad carries, so + // ask for a timestamp that produces this lead against it. + let applied = segment.applied_offset.load(Ordering::Relaxed); + on_output_buffer(state, segment, pad, (input as i64 + lead - applied) as u64); + } + } + + /// A sender that stalls and resumes leaves its stream behind arrival with + /// the offset still right for it. Giving the offset back here would throw + /// the stream forward by that much in a single buffer. + #[test] + fn a_window_spent_behind_arrival_keeps_the_offset() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + let two_seconds = 2 * gst::ClockTime::SECOND.nseconds() as i64; + + window(&anchor.state, &segment, &pad, two_seconds); + assert_eq!( + pad.offset(), + -two_seconds, + "a window spent ahead of arrival is re-anchored" + ); + + window(&anchor.state, &segment, &pad, -two_seconds); + assert_eq!(pad.offset(), -two_seconds, "the offset is left alone"); + } + + /// The caller the offset was measured on has gone. + #[test] + fn a_caller_change_drops_the_offset() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + let two_seconds = 2 * gst::ClockTime::SECOND.nseconds() as i64; + + window(&anchor.state, &segment, &pad, two_seconds); + assert_eq!(pad.offset(), -two_seconds); + + anchor.state.reset(); + // A pad carries the offset it finds until something is pushed through + // it, so the next caller's first buffer is what clears it. + window(&anchor.state, &segment, &pad, 0); + assert_eq!(pad.offset(), 0, "the next caller starts unshifted"); + } +} diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index e6ae73ca4..da90b28e0 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -79,7 +79,7 @@ const LIVE: gst::ClockTime = gst::ClockTime::from_mseconds(2500); /// A 128 kbit/s AAC transport stream in 1316-byte chunks, with each chunk's /// content time, encoded as fast as possible. fn encode_stream() -> Vec<(gst::ClockTime, gst::Buffer)> { - let seconds = (GAP + LIVE + gst::ClockTime::SECOND).seconds() as i32; + let seconds = (GAP + 2 * LIVE + 2 * gst::ClockTime::SECOND).seconds() as i32; let pipeline = gst::parse::launch(&format!( "audiotestsrc num-buffers={} samplesperbuffer=4800 wave=ticks \ ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! avenc_aac bitrate=128000 \ @@ -105,7 +105,7 @@ fn encode_stream() -> Vec<(gst::ClockTime, gst::Buffer)> { } pipeline.set_state(gst::State::Null).unwrap(); assert!( - chunks.last().is_some_and(|(t, _)| *t > GAP + LIVE), + chunks.last().is_some_and(|(t, _)| *t > GAP + 2 * LIVE), "encoder produced too little stream to replay" ); chunks @@ -117,14 +117,17 @@ struct Receiver { leads: Arc>>, } -fn build_receiver(decode: bool, port: u16) -> Receiver { +fn build_receiver(decode: bool, port: u16, keep_listening: bool) -> Receiver { let instance_id = "srt_in"; let mut props: HashMap = HashMap::new(); props.insert("decode".to_string(), PropertyValue::Bool(decode)); props.insert("num_video_tracks".to_string(), PropertyValue::UInt(0)); props.insert("num_audio_tracks".to_string(), PropertyValue::UInt(1)); props.insert("latency".to_string(), PropertyValue::Int(20)); - props.insert("keep_listening".to_string(), PropertyValue::Bool(false)); + props.insert( + "keep_listening".to_string(), + PropertyValue::Bool(keep_listening), + ); props.insert( "srt_uri".to_string(), PropertyValue::String(format!("srt://127.0.0.1:{}?mode=listener", port)), @@ -201,9 +204,8 @@ fn build_receiver(decode: bool, port: u16) -> Receiver { Receiver { pipeline, leads } } -/// Sends the head, then the live part in real time, over one SRT connection. -/// Returns the sender still connected; see [`stop_srt`] for the teardown order. -fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline { +/// One SRT caller, connected and ready to be fed. +fn connect(port: u16) -> (gst::Pipeline, gst_app::AppSrc) { let pipeline = gst::parse::launch(&format!( "appsrc name=in is-live=true format=time \ caps=video/mpegts,systemstream=true,packetsize=188 \ @@ -220,22 +222,46 @@ fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline .downcast::() .unwrap(); pipeline.set_state(gst::State::Playing).unwrap(); + (pipeline, appsrc) +} +/// The stale head, then the tables of the skipped part so the caller's PSI +/// stays continuous across the jump. +fn push_head(appsrc: &gst_app::AppSrc, chunks: &[(gst::ClockTime, gst::Buffer)]) { for (_, buffer) in chunks.iter().filter(|(t, _)| *t < HEAD) { appsrc.push_buffer(buffer.copy()).unwrap(); } for buffer in tables_between(chunks, HEAD, GAP) { appsrc.push_buffer(buffer).unwrap(); } +} + +/// [`LIVE`] of stream from `from`, paced in real time. +fn push_live( + appsrc: &gst_app::AppSrc, + chunks: &[(gst::ClockTime, gst::Buffer)], + from: gst::ClockTime, +) { let start = Instant::now(); - for (t, buffer) in chunks.iter().filter(|(t, _)| *t >= GAP && *t < GAP + LIVE) { - let due = Duration::from_nanos((*t - GAP).nseconds()); + for (t, buffer) in chunks + .iter() + .filter(|(t, _)| *t >= from && *t < from + LIVE) + { + let due = Duration::from_nanos((*t - from).nseconds()); if let Some(wait) = due.checked_sub(start.elapsed()) { std::thread::sleep(wait); } appsrc.push_buffer(buffer.copy()).unwrap(); } std::thread::sleep(Duration::from_millis(300)); +} + +/// Sends the head, then the live part in real time, over one SRT connection. +/// Returns the sender still connected; see [`stop_srt`] for the teardown order. +fn replay(port: u16, chunks: &[(gst::ClockTime, gst::Buffer)]) -> gst::Pipeline { + let (pipeline, appsrc) = connect(port); + push_head(&appsrc, chunks); + push_live(&appsrc, chunks, GAP); pipeline } @@ -313,7 +339,7 @@ fn max_live_lead(decode: bool) -> Option<(i64, usize)> { let chunks = encode_stream(); let port = srt_port(); - let receiver = build_receiver(decode, port); + let receiver = build_receiver(decode, port, false); receiver .pipeline .set_state(gst::State::Playing) @@ -363,6 +389,76 @@ fn assert_on_time(decode: bool) { ); } +/// Whether the caller that follows a corrected one leaves the block on time. +/// +/// `keep_listening` is on by default: one `srtsrc` serves caller after caller +/// and `tsdemux` takes a fresh time reference for each, so a correction that +/// outlives its caller stamps the next one behind its own arrival, where +/// consumers that wait for running time drop it. `second_from` is where the +/// second caller's stream starts: the same encoder carrying on, or a fresh one +/// from zero. +fn assert_second_caller_on_time(second_from: gst::ClockTime) { + gst::init().unwrap(); + if !plugins_available() { + eprintln!("skipping: required GStreamer elements are missing"); + return; + } + + let chunks = encode_stream(); + let port = srt_port(); + let receiver = build_receiver(false, port, true); + receiver + .pipeline + .set_state(gst::State::Playing) + .expect("receiver goes to PLAYING"); + + // A caller with a stale head, so that a correction is made. + let (sender, appsrc) = connect(port); + push_head(&appsrc, &chunks); + push_live(&appsrc, &chunks, GAP); + stop_srt(&sender, "out"); + std::thread::sleep(Duration::from_millis(500)); + let boundary = receiver.leads.lock().unwrap().len(); + + // A second caller, with no stale head of its own. Nothing is replayed + // ahead of it: re-sending tables the first caller already sent breaks the + // continuity counters, which on GStreamer 1.24.2 costs the stream its pad. + let (sender, appsrc) = connect(port); + push_live(&appsrc, &chunks, second_from); + stop_srt(&receiver.pipeline, "srt_in:srtsrc"); + stop_srt(&sender, "out"); + + let leads = receiver.leads.lock().unwrap().clone(); + let first = leads + .get(boundary) + .map(|(i, _)| *i) + .expect("no audio left the block for the second caller, so the test proved nothing"); + // Only the reconnect itself: the correction must go with the caller that + // was measured, not a window or two later. + let settle = first + gst::ClockTime::from_mseconds(300).nseconds(); + let settled: Vec = leads[boundary..] + .iter() + .filter(|(input, _)| *input >= settle) + .map(|(_, lead)| *lead) + .collect(); + assert!( + settled.len() > 20, + "only {} buffers from the second caller after it settled — too few to judge", + settled.len() + ); + let worst = settled + .iter() + .copied() + .max_by_key(|lead| lead.abs()) + .unwrap(); + assert!( + worst.abs() < MAX_LEAD.nseconds() as i64, + "the second caller's audio left the block {} ms off its own arrival ({} buffers). The correction made for the caller that has gone is still on the pads, so every consumer that waits for running time drops this caller as late.", + worst / 1_000_000, + settled.len() + ); +} + #[test] fn stale_head_is_not_stamped_ahead_decoded() { assert_on_time(true); @@ -372,3 +468,13 @@ fn stale_head_is_not_stamped_ahead_decoded() { fn stale_head_is_not_stamped_ahead_passthrough() { assert_on_time(false); } + +#[test] +fn second_caller_restarting_is_not_stamped_behind() { + assert_second_caller_on_time(gst::ClockTime::ZERO); +} + +#[test] +fn second_caller_resuming_is_not_stamped_behind() { + assert_second_caller_on_time(GAP + LIVE); +} From 1a2b88ca9d8a2272733517d8f761973a79c96e2d Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Wed, 23 Sep 2026 14:18:00 -0700 Subject: [PATCH 4/7] test(mpegtssrt_input): run the reconnect tests on 1.24.3 and later only Before 1.24.3 a re-parsed PMT reads as a new program under ignore-pcr, and a caller change re-parses it. tsdemux replaces its source pad, the block does not relink it (#856), and the second caller delivers no audio at all -- so the reconnect tests cannot pass on Ubuntu 24.04 or CI, whatever the anchor does. Skip them below 1.24.3 with the version in the message. They run on the release image, which ships 1.26, and on development machines. CI keeps the two stale-head tests. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/mpegtssrt_input_stale_head_test.rs | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index da90b28e0..be60aa929 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -389,6 +389,18 @@ fn assert_on_time(decode: bool) { ); } +/// Whether this GStreamer keeps the demuxer's pad across a caller change. +/// +/// Before 1.24.3, a re-parsed PMT reads as a new program under `ignore-pcr` +/// ("mpegtsbase: Fix Program equality check"), and the caller change re-parses +/// it. `tsdemux` then replaces its source pad, which the block does not relink +/// (Eyevinn/strom#856), so a second caller delivers nothing whatever the anchor +/// does. Ubuntu 24.04 and CI ship 1.24.2; the release image ships 1.26. +fn second_caller_keeps_its_pad() -> bool { + let (major, minor, micro, _) = gst::version(); + (major, minor, micro) >= (1, 24, 3) +} + /// Whether the caller that follows a corrected one leaves the block on time. /// /// `keep_listening` is on by default: one `srtsrc` serves caller after caller @@ -403,6 +415,16 @@ fn assert_second_caller_on_time(second_from: gst::ClockTime) { eprintln!("skipping: required GStreamer elements are missing"); return; } + if !second_caller_keeps_its_pad() { + eprintln!( + "skipping: GStreamer {}.{}.{} replaces the tsdemux pad when a second caller \ + connects, so no audio reaches the block's output at all", + gst::version().0, + gst::version().1, + gst::version().2 + ); + return; + } let chunks = encode_stream(); let port = srt_port(); From 828ecacf43c4b27982168419cc267167beb805a9 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Fri, 2 Oct 2026 06:49:43 -0700 Subject: [PATCH 5/7] test(mpegtssrt_input): run the reconnect tests on all GStreamer versions #877 relinks the pad tsdemux swaps in when 1.24.2 re-parses the PMT, so a second caller now delivers audio there too. Both reconnect tests pass on 1.24.2 (Ubuntu 24.04, as CI) and fail there when the caller-change reset is removed, so CI guards it. This reverts commit de6cd996028d8d61c039c79036cfd13831a30a19. Co-Authored-By: Claude Opus 5.5 --- .../tests/mpegtssrt_input_stale_head_test.rs | 22 ------------------- 1 file changed, 22 deletions(-) diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index be60aa929..da90b28e0 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -389,18 +389,6 @@ fn assert_on_time(decode: bool) { ); } -/// Whether this GStreamer keeps the demuxer's pad across a caller change. -/// -/// Before 1.24.3, a re-parsed PMT reads as a new program under `ignore-pcr` -/// ("mpegtsbase: Fix Program equality check"), and the caller change re-parses -/// it. `tsdemux` then replaces its source pad, which the block does not relink -/// (Eyevinn/strom#856), so a second caller delivers nothing whatever the anchor -/// does. Ubuntu 24.04 and CI ship 1.24.2; the release image ships 1.26. -fn second_caller_keeps_its_pad() -> bool { - let (major, minor, micro, _) = gst::version(); - (major, minor, micro) >= (1, 24, 3) -} - /// Whether the caller that follows a corrected one leaves the block on time. /// /// `keep_listening` is on by default: one `srtsrc` serves caller after caller @@ -415,16 +403,6 @@ fn assert_second_caller_on_time(second_from: gst::ClockTime) { eprintln!("skipping: required GStreamer elements are missing"); return; } - if !second_caller_keeps_its_pad() { - eprintln!( - "skipping: GStreamer {}.{}.{} replaces the tsdemux pad when a second caller \ - connects, so no audio reaches the block's output at all", - gst::version().0, - gst::version().1, - gst::version().2 - ); - return; - } let chunks = encode_stream(); let port = srt_port(); From 6d22edaaf4b7afcd953bd57e01289465410f91f7 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Tue, 6 Oct 2026 14:13:45 -0700 Subject: [PATCH 6/7] fix(mpegtssrt_input): re-anchor only at connect, and only once A healthy stream runs ahead of arrival by what the sender packs together: multi-frame audio PES and the video burst at connect. Windows of such a stream reached a 205 ms minimum lead, crossed the 200 ms threshold mid-stream, and the shift stepped running time back about 380 ms, so recorders wrote decreasing DTS. A stale head only occurs at connect, so a correction may now be made only in the first 3 s of a caller's input, at most once, and only above 500 ms. A caller change re-arms the watcher. The integration tests keep their own 200 ms on-time bound instead of borrowing MAX_LEAD. Fixes #997 Co-Authored-By: Claude Opus 5.5 --- backend/src/blocks/builtin/tsdemux_anchor.rs | 174 ++++++++++++++---- .../tests/mpegtssrt_input_stale_head_test.rs | 10 +- 2 files changed, 141 insertions(+), 43 deletions(-) diff --git a/backend/src/blocks/builtin/tsdemux_anchor.rs b/backend/src/blocks/builtin/tsdemux_anchor.rs index d788f9637..aae942610 100644 --- a/backend/src/blocks/builtin/tsdemux_anchor.rs +++ b/backend/src/blocks/builtin/tsdemux_anchor.rs @@ -15,24 +15,28 @@ //! so the source is delayed by however long the caller waited to connect. //! //! `srtsrc` stamps each buffer with its arrival time less the SRT transit -//! delay, so healthy demuxed output never runs ahead of the latest input by -//! more than the few frames one SRT message carries. This watcher compares the -//! two, and when the smallest lead over a window exceeds [`MAX_LEAD`] it shifts -//! all of the demuxer's source pads back by the window's mean lead with a pad -//! offset. One correction is shared by every pad, so audio and video stay -//! aligned. +//! delay, so healthy demuxed output runs ahead of the latest input only by what +//! the sender packs together: the frames one SRT message carries, multi-frame +//! audio PES, and a video burst at connect that can hold the lead at 440 ms. +//! This watcher compares the two, and when the smallest lead over a window +//! exceeds [`MAX_LEAD`] it shifts all of the demuxer's source pads back by the +//! window's mean lead with a pad offset. One correction is shared by every pad, +//! so audio and video stay aligned. +//! +//! A stale head only happens at connect, so a correction may only be made in +//! the first [`ARM_PERIOD`] of a caller's input, and only once. Shifting back +//! mid-stream steps running time backwards, which recorders write as +//! decreasing timestamps; a window of a healthy stream that crosses the +//! threshold later on must not be able to do that. //! //! The shift belongs to the caller it was measured on: `srtsrc` keeps listening //! after a caller leaves and `tsdemux` takes a fresh reference for the next one, -//! so the shift is dropped when the caller changes. Output that sits *behind* -//! arrival is left alone, whatever the shift: a sender that stalls and resumes -//! looks the same from here, and giving the shift back would throw its stream -//! forward by that much in one buffer. +//! so the shift is dropped, and the watcher re-armed, when the caller changes. //! //! Timing comes only from buffer timestamps, never the clock: the probes are on //! the streaming thread for every demuxed frame and every SRT message, so they //! use atomics only and do no locking or allocation outside the rare -//! correction. +//! correction. Once disarmed, the output probe only keeps the offset applied. use gstreamer as gst; use gstreamer::prelude::*; @@ -45,14 +49,28 @@ use tracing::info; /// Healthy lead is a sawtooth: every PES completed within one SRT message is /// placed from the same reference, while the input time is that of the whole /// message. A 1316-byte message of 128 kbit/s AAC spans about 100 ms, and lower -/// bitrates span more. -pub const MAX_LEAD: gst::ClockTime = gst::ClockTime::from_mseconds(200); +/// bitrates span more. ffmpeg's default TS muxing packs about 11 AAC frames per +/// PES, which leads by 100-230 ms, and the video burst at connect holds video +/// 290-440 ms ahead; windows of such a stream have reached 205 ms. A stale +/// head leads by however long the caller waited to connect, seconds in +/// practice, so one that waited less than this goes uncorrected. +pub const MAX_LEAD: gst::ClockTime = gst::ClockTime::from_mseconds(500); + +/// Input time after a caller's first buffer within which a correction may be +/// made. The first window closes over the stale head itself, so the correction +/// lands about two windows in. +pub const ARM_PERIOD: gst::ClockTime = gst::ClockTime::from_seconds(3); /// Input time covered by one measurement window. pub const WINDOW: gst::ClockTime = gst::ClockTime::from_mseconds(500); const NONE: u64 = u64::MAX; +/// `arm_deadline` before the caller's first input buffer. +const ARM_PENDING: u64 = u64::MAX; +/// `arm_deadline` once the arm period is over or the correction is made. +const DISARMED: u64 = 0; + struct State { instance_id: String, /// Running time of the most recent input buffer, or `NONE`. @@ -66,16 +84,23 @@ struct State { window_samples: AtomicI64, /// Offset every source pad of the demuxer should carry (ns, never positive). correction: AtomicI64, + /// Input running time after which no correction is made, or `ARM_PENDING` + /// or `DISARMED`. + arm_deadline: AtomicU64, } impl State { - /// Drop the correction and the window in progress. + /// Drop the correction and the window in progress, and re-arm. /// /// The window counters are re-initialised by the first buffer after this, /// which takes the `window_start == NONE` branch of [`on_output_buffer`]. + /// The arm period starts at the next caller's first input buffer, so the + /// input time of the caller that left is forgotten too. fn reset(&self) { self.correction.store(0, Ordering::Relaxed); self.window_start.store(NONE, Ordering::Relaxed); + self.last_input.store(NONE, Ordering::Relaxed); + self.arm_deadline.store(ARM_PENDING, Ordering::Relaxed); } } @@ -96,6 +121,7 @@ impl TsDemuxAnchor { window_lead_sum: AtomicI64::new(0), window_samples: AtomicI64::new(0), correction: AtomicI64::new(0), + arm_deadline: AtomicU64::new(ARM_PENDING), }), } } @@ -108,13 +134,13 @@ impl TsDemuxAnchor { let state = self.state.clone(); pad.add_probe(gst::PadProbeType::BUFFER, move |_, info| { if let Some(pts) = info.buffer().and_then(|b| b.pts()) { - state.last_input.store(pts.nseconds(), Ordering::Relaxed); + on_input_buffer(&state, pts.nseconds()); } gst::PadProbeReturn::Ok }); } - /// Drop the correction whenever the SRT caller changes. + /// Drop the correction and re-arm whenever the SRT caller changes. /// /// Listener mode only: `srtsrc` emits these while accepting and dropping a /// caller, and neither fires in caller mode. @@ -222,19 +248,36 @@ impl PadSegment { } } +fn on_input_buffer(state: &State, ts: u64) { + state.last_input.store(ts, Ordering::Relaxed); + if state.arm_deadline.load(Ordering::Relaxed) == ARM_PENDING { + state + .arm_deadline + .store(ts + ARM_PERIOD.nseconds(), Ordering::Relaxed); + } +} + fn on_output_buffer(state: &State, segment: &PadSegment, pad: &gst::Pad, ts: u64) { let correction = state.correction.load(Ordering::Relaxed); if segment.applied_offset.load(Ordering::Relaxed) != correction { apply(segment, pad, correction); } - let input = state.last_input.load(Ordering::Relaxed); - let Some(running_time) = segment.running_time(ts) else { + let deadline = state.arm_deadline.load(Ordering::Relaxed); + if deadline == DISARMED || deadline == ARM_PENDING { return; - }; + } + let input = state.last_input.load(Ordering::Relaxed); if input == NONE { return; } + if input > deadline { + state.arm_deadline.store(DISARMED, Ordering::Relaxed); + return; + } + let Some(running_time) = segment.running_time(ts) else { + return; + }; let lead = running_time - input as i64; let window_start = state.window_start.load(Ordering::Relaxed); @@ -265,6 +308,7 @@ fn on_output_buffer(state: &State, segment: &PadSegment, pad: &gst::Pad, ts: u64 let mean_lead = lead_sum / samples; let correction = correction - mean_lead; state.correction.store(correction, Ordering::Relaxed); + state.arm_deadline.store(DISARMED, Ordering::Relaxed); apply(segment, pad, correction); info!( "MPEGTSSRT Input {}: demuxed timestamps ran {} ms ahead of arrival, re-anchoring (total offset {} ms)", @@ -298,52 +342,102 @@ mod tests { (pad, segment) } - /// One full window of buffers whose running time sits `lead` from arrival. - fn window(state: &State, segment: &PadSegment, pad: &gst::Pad, lead: i64) { - for step in 0..8u64 { - let input = step * 100 * gst::ClockTime::MSECOND.nseconds(); - state.last_input.store(input, Ordering::Relaxed); + fn ms(ms: i64) -> i64 { + ms * gst::ClockTime::MSECOND.nseconds() as i64 + } + + /// Eight buffers 100 ms of input apart from `start`, enough to close one + /// window, whose running time sits `leads[i % leads.len()]` from arrival. + fn window(state: &State, segment: &PadSegment, pad: &gst::Pad, start: i64, leads: &[i64]) { + for step in 0..8 { + let input = start + ms(step * 100); + on_input_buffer(state, input as u64); + let lead = leads[step as usize % leads.len()]; // running_time(ts) is ts plus whatever offset the pad carries, so // ask for a timestamp that produces this lead against it. let applied = segment.applied_offset.load(Ordering::Relaxed); - on_output_buffer(state, segment, pad, (input as i64 + lead - applied) as u64); + on_output_buffer(state, segment, pad, (input + lead - applied) as u64); + } + } + + /// ffmpeg's multi-frame audio PES and the video burst at connect keep a + /// healthy stream ahead of arrival, and the smallest lead in a window has + /// reached 205 ms. Re-anchoring it steps running time back mid-stream, and + /// recorders write decreasing DTS. + #[test] + fn a_healthy_stream_is_never_re_anchored() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + + // Through the arm period: video at the top of its range, audio at the + // top of its own. + for start in (0..4).map(|i| ms(i * 800)) { + window(&anchor.state, &segment, &pad, start, &[ms(440), ms(230)]); } + assert_eq!(pad.offset(), 0, "the lead at connect is left alone"); + + window( + &anchor.state, + &segment, + &pad, + ms(10_000), + &[ms(440), ms(205)], + ); + assert_eq!(pad.offset(), 0, "a high window mid-stream is left alone"); } - /// A sender that stalls and resumes leaves its stream behind arrival with - /// the offset still right for it. Giving the offset back here would throw - /// the stream forward by that much in a single buffer. + /// A stale head is corrected once, early in the connection. #[test] - fn a_window_spent_behind_arrival_keeps_the_offset() { + fn a_stale_head_is_re_anchored_once() { let anchor = TsDemuxAnchor::new("test"); let (pad, segment) = pad_and_segment(); - let two_seconds = 2 * gst::ClockTime::SECOND.nseconds() as i64; - window(&anchor.state, &segment, &pad, two_seconds); + window(&anchor.state, &segment, &pad, 0, &[ms(2000)]); assert_eq!( pad.offset(), - -two_seconds, + -ms(2000), "a window spent ahead of arrival is re-anchored" ); - window(&anchor.state, &segment, &pad, -two_seconds); - assert_eq!(pad.offset(), -two_seconds, "the offset is left alone"); + window(&anchor.state, &segment, &pad, ms(800), &[ms(1000)]); + assert_eq!(pad.offset(), -ms(2000), "a second correction is not made"); + } + + /// Past the arm period even a large lead is left alone: shifting back + /// there would step every recording's timestamps backwards. + #[test] + fn no_correction_after_the_arm_period() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + + window(&anchor.state, &segment, &pad, 0, &[0]); + let late = ms(ARM_PERIOD.mseconds() as i64 + 1000); + window(&anchor.state, &segment, &pad, late, &[ms(2000)]); + assert_eq!(pad.offset(), 0); } /// The caller the offset was measured on has gone. #[test] - fn a_caller_change_drops_the_offset() { + fn a_caller_change_drops_the_offset_and_re_arms() { let anchor = TsDemuxAnchor::new("test"); let (pad, segment) = pad_and_segment(); - let two_seconds = 2 * gst::ClockTime::SECOND.nseconds() as i64; - window(&anchor.state, &segment, &pad, two_seconds); - assert_eq!(pad.offset(), -two_seconds); + window(&anchor.state, &segment, &pad, 0, &[ms(2000)]); + assert_eq!(pad.offset(), -ms(2000)); anchor.state.reset(); // A pad carries the offset it finds until something is pushed through - // it, so the next caller's first buffer is what clears it. - window(&anchor.state, &segment, &pad, 0); + // it, so the next caller's first buffer is what clears it. The next + // caller connects long after the first one's arm period. + window(&anchor.state, &segment, &pad, ms(60_000), &[0]); assert_eq!(pad.offset(), 0, "the next caller starts unshifted"); + + anchor.state.reset(); + window(&anchor.state, &segment, &pad, ms(120_000), &[ms(1500)]); + assert_eq!( + pad.offset(), + -ms(1500), + "a later caller's stale head is corrected" + ); } } diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index da90b28e0..610d0c11a 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -18,7 +18,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use strom::blocks::builtin::mpegtssrt_input::MpegTsSrtInputBuilder; -use strom::blocks::builtin::tsdemux_anchor::{MAX_LEAD, WINDOW}; +use strom::blocks::builtin::tsdemux_anchor::WINDOW; use strom::blocks::{BlockBuildContext, BlockBuilder}; use strom_types::PropertyValue; @@ -69,6 +69,10 @@ fn srt_port() -> u16 { .expect("no free UDP port for the SRT listener") } +/// Largest lead from arrival a buffer may have and still count as on time: +/// the sawtooth of 128 kbit/s AAC in 1316-byte SRT messages, with margin. +const ON_TIME: gst::ClockTime = gst::ClockTime::from_mseconds(200); + /// Where the live part of the replayed stream starts. const GAP: gst::ClockTime = gst::ClockTime::from_seconds(2); /// Stale frames sent ahead of the live part. @@ -379,7 +383,7 @@ fn assert_on_time(decode: bool) { return; }; assert!( - max_lead < MAX_LEAD.nseconds() as i64, + max_lead < ON_TIME.nseconds() as i64, "decode={}: live audio left the block {} ms ahead of its SRT arrival \ ({} buffers). tsdemux took its time reference from the stale head, so \ every sink downstream holds this source back by the caller's wait.", @@ -452,7 +456,7 @@ fn assert_second_caller_on_time(second_from: gst::ClockTime) { .max_by_key(|lead| lead.abs()) .unwrap(); assert!( - worst.abs() < MAX_LEAD.nseconds() as i64, + worst.abs() < ON_TIME.nseconds() as i64, "the second caller's audio left the block {} ms off its own arrival ({} buffers). The correction made for the caller that has gone is still on the pads, so every consumer that waits for running time drops this caller as late.", worst / 1_000_000, settled.len() From 6f6648a908b248a45387f304032df0deaf1b19e5 Mon Sep 17 00:00:00 2001 From: Peter Wagenet Date: Tue, 6 Oct 2026 14:29:06 -0700 Subject: [PATCH 7/7] test(mpegtssrt_input): read TS packets with as_chunks Rust 1.98's clippy denies chunks_exact with a constant size. Co-Authored-By: Claude Opus 5.5 --- backend/tests/mpegtssrt_input_stale_head_test.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/backend/tests/mpegtssrt_input_stale_head_test.rs b/backend/tests/mpegtssrt_input_stale_head_test.rs index 610d0c11a..6b808b3d2 100644 --- a/backend/tests/mpegtssrt_input_stale_head_test.rs +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -305,7 +305,12 @@ fn tables_between( let mut media_pids = std::collections::HashSet::new(); for (_, buffer) in chunks { let map = buffer.map_readable().unwrap(); - for packet in map.chunks_exact(TS_PACKET).filter(|p| starts_pes(p)) { + for packet in map + .as_chunks::() + .0 + .iter() + .filter(|p| starts_pes(*p)) + { media_pids.insert(ts_pid(packet)); } } @@ -315,8 +320,10 @@ fn tables_between( .filter_map(|(_, buffer)| { let map = buffer.map_readable().unwrap(); let tables: Vec = map - .chunks_exact(TS_PACKET) - .filter(|p| !media_pids.contains(&ts_pid(p))) + .as_chunks::() + .0 + .iter() + .filter(|p| !media_pids.contains(&ts_pid(*p))) .flatten() .copied() .collect();