Fix fMP4 zero-duration samples

rs/moq-mux/src/container/fmp4/export.rs

@@ -9,8 +9,8 @@ use mp4_atom::{DecodeMaybe, Encode};
 use crate::Result;
 use crate::catalog::Stream;
 use crate::container::ExportSource;
-use crate::container::Frame;
 use crate::container::fmp4::Error;
+use crate::container::{Frame, Timestamp};
 
 /// Subscribe to a moq broadcast and produce a single fMP4 / CMAF byte stream.
 ///
@@ -249,7 +249,7 @@ impl<S: Stream> Export<S> {
 			let flush_before = should_flush(track, &frame, frag, has_video_track);
 			if flush_before {
 				let frames = std::mem::take(&mut track.buffer);
-				let fragment = emit_fragment(track, frames)?;
+				let fragment = emit_fragment(track, frames, Some(&frame))?;
 				// The flushed run is done; the incoming frame opens the next buffer.
 				track.buffer_independent = frame.keyframe;
 				track.buffer.push(frame);
@@ -281,7 +281,7 @@ impl<S: Stream> Export<S> {
 		if let Some(name) = flushable {
 			let track = self.tracks.get_mut(&name).unwrap();
 			let frames = std::mem::take(&mut track.buffer);
-			let fragment = emit_fragment(track, frames)?;
+			let fragment = emit_fragment(track, frames, None)?;
 			return Poll::Ready(Ok(Some(fragment)));
 		}
 
@@ -556,10 +556,11 @@ fn encode_fragment(track: &mut Fmp4Track, frames: Vec<Frame>) -> Result<Bytes> {
 }
 
 /// Encode a buffered run and wrap it with the metadata a segmenting consumer needs.
-fn emit_fragment(track: &mut Fmp4Track, frames: Vec<Frame>) -> Result<Fragment> {
+fn emit_fragment(track: &mut Fmp4Track, frames: Vec<Frame>, successor: Option<&Frame>) -> Result<Fragment> {
 	// Audio has no keyframes, so every audio fragment is independent; video is
 	// independent only when its buffer opened on a keyframe (a GOP boundary).
 	let independent = !track.is_video || track.buffer_independent;
+	let frames = infer_missing_durations(frames, successor, track.default_frame);
 	let duration = fragment_seconds(&frames, track.default_frame);
 	let data = encode_fragment(track, frames)?;
 	Ok(Fragment {
@@ -580,7 +581,10 @@ fn fragment_seconds(frames: &[Frame], default_frame: Duration) -> f64 {
 	if frames.is_empty() {
 		return 0.0;
 	}
-	if frames.iter().all(|f| f.duration.is_some()) {
+	if frames
+		.iter()
+		.all(|f| f.duration.is_some_and(|duration| !duration.is_zero()))
+	{
 		return frames
 			.iter()
 			.map(|f| Duration::from(f.duration.unwrap()))
@@ -597,6 +601,42 @@ fn fragment_seconds(frames: &[Frame], default_frame: Duration) -> f64 {
 	((max - min) + default_frame).as_secs_f64()
 }
 
+fn infer_missing_durations(mut frames: Vec<Frame>, successor: Option<&Frame>, default_frame: Duration) -> Vec<Frame> {
+	let infer_from_pts = pts_monotonic(&frames, successor);
+	let fallback = Timestamp::try_from(default_frame).ok();
+
+	for i in 0..frames.len() {
+		if frames[i].duration.is_some_and(|duration| !duration.is_zero()) {
+			continue;
+		}
+
+		frames[i].duration = infer_from_pts
+			.then(|| next_timestamp(&frames, successor, i))
+			.flatten()
+			.and_then(|timestamp| timestamp.checked_sub(frames[i].timestamp).ok())
+			.filter(|duration| !duration.is_zero())
+			.or(fallback);
+	}
+
+	frames
+}
+
+fn pts_monotonic(frames: &[Frame], successor: Option<&Frame>) -> bool {
+	let frames_monotonic = frames.windows(2).all(|pair| pair[1].timestamp >= pair[0].timestamp);
+	let successor_monotonic = match (frames.last(), successor) {
+		(Some(last), Some(successor)) => successor.timestamp >= last.timestamp,
+		_ => true,
+	};
+	frames_monotonic && successor_monotonic
+}
+
+fn next_timestamp(frames: &[Frame], successor: Option<&Frame>, index: usize) -> Option<Timestamp> {
+	frames
+		.get(index + 1)
+		.map(|next| next.timestamp)
+		.or_else(|| successor.map(|next| next.timestamp))
+}
+
 fn catalog_timescale_video(config: &VideoConfig) -> u64 {
 	match &config.container {
 		Container::Cmaf { init, .. } => {
@@ -623,3 +663,69 @@ fn parse_timescale_from_init(init: &[u8]) -> Result<u64> {
 	}
 	Err(Error::NoMoov.into())
 }
+
+#[cfg(test)]
+mod tests {
+	use bytes::Bytes;
+
+	use super::*;
+
+	fn ts(micros: u64) -> Timestamp {
+		Timestamp::from_micros(micros).unwrap()
+	}
+
+	fn frame(timestamp_us: u64, duration_us: Option<u64>) -> Frame {
+		Frame {
+			timestamp: ts(timestamp_us),
+			duration: duration_us.map(ts),
+			payload: Bytes::from_static(&[0xDE, 0xAD]),
+			keyframe: false,
+		}
+	}
+
+	#[test]
+	fn infer_missing_durations_uses_default_for_trailing_sample() {
+		let frames = infer_missing_durations(
+			vec![frame(0, Some(0)), frame(41_667, None), frame(83_334, None)],
+			None,
+			Duration::from_millis(33),
+		);
+
+		assert_eq!(frames[0].duration, Some(ts(41_667)));
+		assert_eq!(frames[1].duration, Some(ts(41_667)));
+		assert_eq!(frames[2].duration, Some(ts(33_000)));
+		assert_eq!(fragment_seconds(&frames, Duration::from_millis(33)), 0.116334);
+	}
+
+	#[test]
+	fn infer_missing_duration_uses_default_for_single_frame() {
+		let frames = infer_missing_durations(vec![frame(83_333, Some(0))], None, Duration::from_millis(40));
+
+		assert_eq!(frames[0].duration, Some(ts(40_000)));
+		assert_eq!(fragment_seconds(&frames, Duration::from_millis(40)), 0.04);
+	}
+
+	#[test]
+	fn infer_trailing_duration_from_successor_frame() {
+		let successor = frame(83_334, None);
+		let frames = infer_missing_durations(vec![frame(41_667, None)], Some(&successor), Duration::from_millis(33));
+
+		assert_eq!(frames[0].duration, Some(ts(41_667)));
+		assert_eq!(fragment_seconds(&frames, Duration::from_millis(33)), 0.041667);
+	}
+
+	#[test]
+	fn infer_missing_durations_avoids_non_monotonic_pts() {
+		let successor = frame(66_000, None);
+		let frames = infer_missing_durations(
+			vec![frame(0, None), frame(99_000, None), frame(33_000, None)],
+			Some(&successor),
+			Duration::from_millis(33),
+		);
+
+		assert_eq!(frames[0].duration, Some(ts(33_000)));
+		assert_eq!(frames[1].duration, Some(ts(33_000)));
+		assert_eq!(frames[2].duration, Some(ts(33_000)));
+		assert_eq!(fragment_seconds(&frames, Duration::from_millis(33)), 0.099);
+	}
+}

rs/moq-mux/src/container/fmp4/import.rs

@@ -443,6 +443,8 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 			let mut min_timestamp = None;
 			let mut max_timestamp = None;
 			let mut contains_keyframe = false;
+			let total_samples: usize = traf.trun.iter().map(|t| t.entries.len()).sum();
+			let mut sample_index = 0usize;
 
 			for trun in &traf.trun {
 				let tfhd = &traf.tfhd;
@@ -472,11 +474,17 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 						.unwrap_or(tfhd.default_sample_flags.unwrap_or(default_sample_flags));
 					let duration = entry
 						.duration
-						.unwrap_or(tfhd.default_sample_duration.unwrap_or(default_sample_duration));
+						.or(tfhd.default_sample_duration)
+						.or(Some(default_sample_duration))
+						.filter(|duration| *duration != 0);
 					let size = entry
 						.size
 						.unwrap_or(tfhd.default_sample_size.unwrap_or(default_sample_size)) as usize;
 
+					if duration.is_none() && sample_index + 1 != total_samples {
+						return Err(Error::MissingSampleDuration.into());
+					}
+
 					// Checked: a negative composition offset must not wrap into a huge u64 PTS.
 					let pts = dts
 						.checked_add_signed(entry.cts.unwrap_or_default() as i64)
@@ -517,8 +525,11 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 
 					track.last_timestamp = Some(timestamp);
 
-					dts = dts.checked_add(duration as u64).ok_or(Error::PtsOverflow)?;
+					if let Some(duration) = duration {
+						dts = dts.checked_add(duration as u64).ok_or(Error::PtsOverflow)?;
+					}
 					offset = sample_end;
+					sample_index += 1;
 				}
 			}
 
@@ -555,9 +566,17 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 			// and ensuring trun.data_offset is Some(...) reserves 4 bytes per trun.
 			for traf_mut in &mut adjusted_moof.traf {
 				traf_mut.tfhd.base_data_offset = None;
+				if traf_mut.tfhd.default_sample_duration == Some(0) {
+					traf_mut.tfhd.default_sample_duration = None;
+				}
 				for trun_mut in &mut traf_mut.trun {
 					// Reserve the data_offset field; the real value is filled in below.
 					trun_mut.data_offset = Some(0);
+					for entry in &mut trun_mut.entries {
+						if entry.duration == Some(0) {
+							entry.duration = None;
+						}
+					}
 				}
 			}
 

rs/moq-mux/src/container/fmp4/mod.rs

@@ -262,7 +262,7 @@ pub(crate) fn decode(data: Bytes, timescale: u64) -> Result<Vec<Frame>> {
 
 			// Carry the sample-duration through at the track's scale when present, so
 			// the jitter buffer can use it and an exporter can write it back.
-			let sample_duration = entry.duration.or(default_duration);
+			let sample_duration = entry.duration.or(default_duration).filter(|d| *d != 0);
 
 			// The last sample needs no duration (nothing follows it to time), but any
 			// earlier sample without one makes the rest of the fragment's DTS ambiguous.
@@ -694,4 +694,43 @@ mod tests {
 		assert_eq!(frames.len(), 1);
 		assert_eq!(frames[0].duration, None);
 	}
+
+	#[test]
+	fn decode_zero_duration_reports_none() {
+		use mp4_atom::Encode;
+
+		let timescale = 24_000;
+		let moof = mp4_atom::Moof {
+			mfhd: mp4_atom::Mfhd { sequence_number: 0 },
+			traf: vec![mp4_atom::Traf {
+				tfhd: mp4_atom::Tfhd {
+					track_id: 1,
+					default_sample_duration: Some(0),
+					default_sample_size: Some(2),
+					..Default::default()
+				},
+				tfdt: Some(mp4_atom::Tfdt {
+					base_media_decode_time: 2_000,
+				}),
+				trun: vec![mp4_atom::Trun {
+					data_offset: Some(0),
+					entries: vec![mp4_atom::TrunEntry {
+						size: None,
+						duration: None,
+						..Default::default()
+					}],
+				}],
+				..Default::default()
+			}],
+		};
+
+		let mut buf = Vec::new();
+		moof.encode(&mut buf).unwrap();
+		mp4_atom::Mdat { data: vec![0xDE, 0xAD] }.encode(&mut buf).unwrap();
+
+		let frames = decode(Bytes::from(buf), timescale).unwrap();
+		assert_eq!(frames.len(), 1);
+		assert_eq!(frames[0].timestamp.as_micros(), 83_333);
+		assert_eq!(frames[0].duration, None);
+	}
 }