fix(moq-gst): stop moqsrc panicking on backwards timestamps; globally unique pad ids

rs/moq-gst/Cargo.toml

@@ -13,7 +13,6 @@ publish = true
 [lib]
 name = "gstmoq"
 crate-type = ["cdylib", "rlib"]
-test = false
 doctest = false
 
 [dependencies]

rs/moq-gst/src/source/imp.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{LazyLock, Mutex};
 use std::time::Duration;
 
@@ -20,6 +21,10 @@ static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
 		.expect("spawn tokio runtime")
 });
 
+/// Process-wide pad id counter. Kept global (not per-session) so a pad created by a
+/// restarted session can't collide with one still being torn down by the previous one.
+static NEXT_PAD_ID: AtomicU64 = AtomicU64::new(0);
+
 #[derive(Debug, Clone, Default)]
 struct Settings {
 	url: Option<String>,
@@ -289,7 +294,6 @@ async fn run_session(
 	// disappearing, or changing codec/resolution mid-stream.
 	let mut active: HashMap<String, ActiveTrack> = HashMap::new();
 	let mut pumps: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
-	let mut next_pad_id: u64 = 0;
 	let mut catalog_closed = false;
 
 	loop {
@@ -319,7 +323,7 @@ async fn run_session(
 			// returning None) while we wait for the remaining pumps to drain.
 			next = catalog_consumer.next(), if !catalog_closed => {
 				match next? {
-					Some(catalog) => reconcile(&catalog, &mut active, &mut pumps, &mut next_pad_id, &broadcast, &element)?,
+					Some(catalog) => reconcile(&catalog, &mut active, &mut pumps, &broadcast, &element)?,
 					// Catalog track closed. Don't cancel the pumps: let each reach its
 					// natural Ok(None) -> EOS end so downstream sees a clean EOS rather than a
 					// bare pad drop. We just stop reconciling and wait for them to drain.
@@ -338,7 +342,6 @@ fn reconcile(
 	catalog: &hang::catalog::Catalog,
 	active: &mut HashMap<String, ActiveTrack>,
 	pumps: &mut tokio::task::JoinSet<()>,
-	next_pad_id: &mut u64,
 	broadcast: &moq_net::BroadcastConsumer,
 	element: &glib::WeakRef<super::MoqSrc>,
 ) -> Result<()> {
@@ -397,8 +400,7 @@ fn reconcile(
 			continue;
 		}
 
-		let id = *next_pad_id;
-		*next_pad_id += 1;
+		let id = NEXT_PAD_ID.fetch_add(1, Ordering::Relaxed);
 
 		let track_consumer = broadcast.subscribe_track(&moq_net::Track::new(&name))?;
 		let track = moq_mux::container::Consumer::new(track_consumer, d.container).with_latency(Duration::from_secs(1));
@@ -528,10 +530,7 @@ fn build_buffer(
 	let buffer_mut = buffer.get_mut().unwrap();
 
 	let pts = match *reference_ts {
-		Some(reference) => {
-			let delta: Duration = (frame.timestamp - reference).into();
-			gst::ClockTime::from_nseconds(delta.as_nanos() as u64)
-		}
+		Some(reference) => relative_pts(frame.timestamp, reference),
 		None => {
 			*reference_ts = Some(frame.timestamp);
 			gst::ClockTime::ZERO
@@ -551,6 +550,18 @@ fn build_buffer(
 	buffer
 }
 
+/// PTS of `timestamp` relative to the track's first frame (`reference`).
+///
+/// Frames arrive in decode order, so a B-frame's presentation timestamp can fall before
+/// the reference. `Timestamp` subtraction panics on underflow, so clamp to zero rather
+/// than crash the pump (which would leak its pad).
+fn relative_pts(timestamp: moq_mux::container::Timestamp, reference: moq_mux::container::Timestamp) -> gst::ClockTime {
+	match timestamp.checked_sub(reference) {
+		Ok(delta) => gst::ClockTime::from_nseconds(Duration::from(delta).as_nanos() as u64),
+		Err(_) => gst::ClockTime::ZERO,
+	}
+}
+
 fn video_caps(config: &hang::catalog::VideoConfig) -> Result<gst::Caps> {
 	use hang::catalog::VideoCodec;
 
@@ -629,3 +640,28 @@ fn audio_caps(config: &hang::catalog::AudioConfig) -> Result<gst::Caps> {
 	};
 	Ok(caps)
 }
+
+#[cfg(test)]
+mod tests {
+	use super::relative_pts;
+	use moq_mux::container::Timestamp;
+
+	#[test]
+	fn relative_pts_clamps_backwards_timestamps() {
+		let reference = Timestamp::from_millis(2000).unwrap();
+
+		// A frame presenting before the reference (a decode-order B-frame) must clamp to
+		// zero, not underflow and panic.
+		assert_eq!(
+			relative_pts(Timestamp::from_millis(1000).unwrap(), reference),
+			gst::ClockTime::ZERO
+		);
+		assert_eq!(relative_pts(reference, reference), gst::ClockTime::ZERO);
+
+		// A forward timestamp yields the delta.
+		assert_eq!(
+			relative_pts(Timestamp::from_millis(2500).unwrap(), reference),
+			gst::ClockTime::from_mseconds(500)
+		);
+	}
+}