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..6d79bf169 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,20 @@ 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); + } + anchor.watch_caller(&srtsrc); + 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 +247,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 +257,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 +280,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..aae942610 --- /dev/null +++ b/backend/src/blocks/builtin/tsdemux_anchor.rs @@ -0,0 +1,443 @@ +//! 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 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, 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. Once disarmed, the output probe only keeps the offset applied. + +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. 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`. + 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, + /// 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, 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); + } +} + +/// 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), + arm_deadline: AtomicU64::new(ARM_PENDING), + }), + } + } + + /// 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()) { + on_input_buffer(&state, pts.nseconds()); + } + gst::PadProbeReturn::Ok + }); + } + + /// 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. + 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. + 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_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 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); + 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); + 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)", + 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); +} + +#[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) + } + + 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 + 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 stale head is corrected once, early in the connection. + #[test] + fn a_stale_head_is_re_anchored_once() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + + window(&anchor.state, &segment, &pad, 0, &[ms(2000)]); + assert_eq!( + pad.offset(), + -ms(2000), + "a window spent ahead of arrival is re-anchored" + ); + + 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_and_re_arms() { + let anchor = TsDemuxAnchor::new("test"); + let (pad, segment) = pad_and_segment(); + + 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. 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 new file mode 100644 index 000000000..6b808b3d2 --- /dev/null +++ b/backend/tests/mpegtssrt_input_stale_head_test.rs @@ -0,0 +1,491 @@ +//! 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::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") +} + +/// 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. +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 + 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 \ + ! 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 + 2 * 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, 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(keep_listening), + ); + 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 } +} + +/// 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 \ + ! 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(); + (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 >= 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 +} + +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 + .as_chunks::() + .0 + .iter() + .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 + .as_chunks::() + .0 + .iter() + .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 + .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, false); + 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 < 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.", + decode, + max_lead / 1_000_000, + count + ); +} + +/// 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() < 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() + ); +} + +#[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); +} + +#[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); +}