refactor: carry moq-video decode timestamps as moq_net::Timestamp

rs/libmoq/src/video.rs

@@ -140,7 +140,9 @@ impl Video {
 			// Flatten to CPU bytes outside the lock (a GPU frame downloads here),
 			// then hold the lock only to buffer it; release before the callback.
 			let frame = VideoFrame {
-				timestamp_us: frame.timestamp_us,
+				// The C ABI carries microseconds; the decoded frame's Timestamp is
+				// constrained to a QUIC VarInt, so the microsecond value fits a u64.
+				timestamp_us: frame.timestamp.as_micros() as u64,
 				width: frame.width,
 				height: frame.height,
 				data: frame.into_i420()?,

rs/moq-transcode/src/rung.rs

@@ -253,7 +253,11 @@ async fn transcode_group_inner(
 	demand: Option<&moq_net::track::Demand>,
 ) -> Result<bool, Error> {
 	let mut first = true;
-	let mut last_timestamp = 0u64;
+	// The latest presentation time seen, tracked so a one-shot group can stamp any
+	// packets the encoder still holds at the end. `None` until the first frame:
+	// `Timestamp` compares by raw value at a fixed scale, so there's no scale-neutral
+	// zero to seed it with (the source's scale isn't known until a frame arrives).
+	let mut last_timestamp: Option<moq_net::Timestamp> = None;
 
 	loop {
 		let frames = match demand {
@@ -268,12 +272,9 @@ async fn transcode_group_inner(
 		};
 
 		for frame in frames {
-			let timestamp: u64 = frame
-				.timestamp
-				.as_micros()
-				.try_into()
-				.map_err(|_| moq_net::TimeOverflow)?;
-			last_timestamp = last_timestamp.max(timestamp);
+			let timestamp = frame.timestamp;
+			// All source frames share one scale, so this `max` never crosses scales.
+			last_timestamp = Some(last_timestamp.map_or(timestamp, |last| last.max(timestamp)));
 
 			// A group opens on a keyframe by construction, so the first frame is
 			// an IDR. The low-level `Container::read` the transcoder uses does not
@@ -289,20 +290,18 @@ async fn transcode_group_inner(
 		}
 	}
 
-	if demand.is_none() {
-		// One-shot group: drain whatever the encoder still buffers.
+	if let (None, Some(last_timestamp)) = (demand, last_timestamp) {
+		// One-shot group: drain whatever the encoder still buffers, stamping it with
+		// the last presentation time we saw. No frames read means nothing to drain.
 		write(output, pipeline.finish(last_timestamp)?)?;
 	}
 	Ok(true)
 }
 
 /// Append encoded packets to the output group in the legacy hang framing.
-fn write(output: &mut moq_net::group::Producer, packets: Vec<(u64, Bytes)>) -> Result<(), Error> {
+fn write(output: &mut moq_net::group::Producer, packets: Vec<(moq_net::Timestamp, Bytes)>) -> Result<(), Error> {
 	for (timestamp, payload) in packets {
-		let frame = hang::container::Frame {
-			timestamp: moq_net::Timestamp::from_micros(timestamp)?,
-			payload,
-		};
+		let frame = hang::container::Frame { timestamp, payload };
 		frame.encode(output)?;
 	}
 	Ok(())
@@ -348,11 +347,16 @@ impl Pipeline {
 	}
 
 	/// Transcode one container payload into zero or more encoded packets, each
-	/// paired with its presentation timestamp in microseconds.
-	fn process(&mut self, payload: &Bytes, timestamp: u64, keyframe: bool) -> Result<Vec<(u64, Bytes)>, Error> {
+	/// paired with its presentation timestamp.
+	fn process(
+		&mut self,
+		payload: &Bytes,
+		timestamp: moq_net::Timestamp,
+		keyframe: bool,
+	) -> Result<Vec<(moq_net::Timestamp, Bytes)>, Error> {
 		let mut packets = Vec::new();
 		for raw in self.decoder.decode(payload, timestamp, keyframe)? {
-			let raw_timestamp = raw.timestamp_us;
+			let raw_timestamp = raw.timestamp;
 			let encoded = if (raw.width, raw.height) == (self.width, self.height) {
 				// Already at the rung size (the decoder scaled): feed the frame
 				// through as-is, keeping a GPU frame on the GPU.
@@ -370,7 +374,7 @@ impl Pipeline {
 	}
 
 	/// Drain the encoder, pairing any buffered packets with `timestamp`.
-	fn finish(&mut self, timestamp: u64) -> Result<Vec<(u64, Bytes)>, Error> {
+	fn finish(&mut self, timestamp: moq_net::Timestamp) -> Result<Vec<(moq_net::Timestamp, Bytes)>, Error> {
 		Ok(self
 			.encoder
 			.finish()?

rs/moq-video/src/decode/backend/mediafoundation.rs

@@ -29,6 +29,7 @@ use std::mem::ManuallyDrop;
 use std::ptr;
 
 use bytes::Bytes;
+use moq_net::Timestamp;
 use windows::Win32::Graphics::Direct3D11::{ID3D11Device, ID3D11Texture2D};
 use windows::Win32::Media::MediaFoundation::{
 	IMF2DBuffer, IMFDXGIBuffer, IMFDXGIDeviceManager, IMFMediaType, IMFSample, IMFTransform, MF_E_NO_MORE_TYPES,
@@ -353,7 +354,7 @@ impl MediaFoundation {
 }
 
 impl Backend for MediaFoundation {
-	fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
+	fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
 		let mut out = Vec::new();
 
 		let sample = self.build_sample(&access_unit)?;
@@ -383,7 +384,7 @@ impl Backend for MediaFoundation {
 		Ok(out
 			.into_iter()
 			.map(|i420| Decoded {
-				timestamp_us,
+				timestamp,
 				frame: Frame::I420(i420),
 			})
 			.collect())

rs/moq-video/src/decode/backend/mod.rs

@@ -15,6 +15,7 @@
 //! fallback below the hardware path.
 
 use bytes::Bytes;
+use moq_net::Timestamp;
 
 use super::decoder::{Config, Kind};
 use crate::Error;
@@ -50,10 +51,10 @@ impl Codec {
 
 /// One decoded picture: the raw frame plus its presentation timestamp.
 pub(crate) struct Decoded {
-	/// Presentation timestamp in microseconds. Backends that decode one-in
-	/// one-out echo the input timestamp; NVDEC threads timestamps through its
-	/// parser, so they survive decoder delay and frame reordering.
-	pub timestamp_us: u64,
+	/// Presentation timestamp. Backends that decode one-in one-out echo the input
+	/// timestamp; NVDEC threads timestamps through its parser, so they survive
+	/// decoder delay and frame reordering.
+	pub timestamp: Timestamp,
 	/// The decoded picture: CPU I420, or a GPU frame the encode side can consume
 	/// without a CPU round trip.
 	pub frame: Frame,
@@ -63,11 +64,10 @@ pub(crate) struct Decoded {
 /// or more decoded frames (zero while the decoder is still buffering, e.g. before
 /// the first keyframe's parameter sets).
 pub(crate) trait Backend: Send {
-	/// Decode one Annex-B access unit stamped with its presentation time in
-	/// microseconds. `keyframe` marks a keyframe (parameter sets are inline ahead
-	/// of it). Takes an owned [`Bytes`] so a backend can split out NAL slices
-	/// without copying.
-	fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, keyframe: bool) -> Result<Vec<Decoded>, Error>;
+	/// Decode one Annex-B access unit stamped with its presentation `timestamp`.
+	/// `keyframe` marks a keyframe (parameter sets are inline ahead of it). Takes an
+	/// owned [`Bytes`] so a backend can split out NAL slices without copying.
+	fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, keyframe: bool) -> Result<Vec<Decoded>, Error>;
 
 	/// The decoder name in use, e.g. `"videotoolbox"` (for logging).
 	fn name(&self) -> &str;

rs/moq-video/src/decode/backend/nvdec.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use cudarc::driver::CudaContext;
+use moq_net::Timestamp;
 use moq_nvenc::cuvid;
 use moq_nvenc::sys::cuviddec::{
 	CUVIDDECODECAPS, CUVIDDECODECREATEINFO, CUVIDPICPARAMS, CUVIDPROCPARAMS, CUvideodecoder, cudaVideoChromaFormat,
@@ -161,7 +162,7 @@ impl Nvdec {
 }
 
 impl Backend for Nvdec {
-	fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
+	fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
 		// The parser callbacks (decoder create, decode) and the map/copy below
 		// all need the CUDA context current on this thread.
 		self.state
@@ -177,7 +178,8 @@ impl Backend for Nvdec {
 				| (CUvideopacketflags::CUVID_PKT_ENDOFPICTURE as c_ulong),
 			payload_size: access_unit.len() as c_ulong,
 			payload: access_unit.as_ptr(),
-			timestamp: timestamp_us as i64,
+			// cuvid's clock rate is microseconds (set at parser creation).
+			timestamp: timestamp.as_micros() as i64,
 		};
 
 		// SAFETY: parser and packet are valid; the payload outlives the call
@@ -384,7 +386,7 @@ impl State {
 
 		Ok(Decoded {
 			// Timestamps rode the parser from `decode` (microseconds, unsigned).
-			timestamp_us: disp.timestamp.max(0) as u64,
+			timestamp: Timestamp::from_micros(disp.timestamp.max(0) as u64).unwrap_or(Timestamp::ZERO),
 			frame: Frame::Cuda(copied?),
 		})
 	}
@@ -512,10 +514,11 @@ mod tests {
 		let rgba = gradient_rgba(w, h);
 		let mut out = Vec::new();
 		for i in 0..10u64 {
+			let timestamp = Timestamp::from_micros(i * 33_333).unwrap();
 			for packet in encoder.encode_rgba(&rgba, w, h, i == 0).unwrap() {
-				for decoded in decoder.decode(packet, i * 33_333, i == 0).unwrap() {
+				for decoded in decoder.decode(packet, timestamp, i == 0).unwrap() {
 					let i420 = decoded.frame.to_i420().unwrap().into_owned();
-					out.push((decoded.timestamp_us, i420));
+					out.push((decoded.timestamp.as_micros() as u64, i420));
 				}
 			}
 		}
@@ -646,8 +649,9 @@ mod tests {
 			let mut decoder = decoder;
 			let mut frames = Vec::new();
 			for i in 0..10u64 {
+				let timestamp = Timestamp::from_micros(i * 33_333).unwrap();
 				for packet in source.encode_rgba(&rgba, w, h, i == 0).unwrap() {
-					frames.extend(decoder.decode(packet, i * 33_333, i == 0).unwrap());
+					frames.extend(decoder.decode(packet, timestamp, i == 0).unwrap());
 				}
 			}
 			frames
@@ -686,7 +690,10 @@ mod tests {
 		.unwrap();
 		let mut last = None;
 		for (i, packet) in packets.into_iter().enumerate() {
-			for out in check.decode(packet, i as u64, i == 0).unwrap() {
+			for out in check
+				.decode(packet, Timestamp::from_micros(i as u64).unwrap(), i == 0)
+				.unwrap()
+			{
 				last = Some(out.frame.to_i420().unwrap().into_owned());
 			}
 		}

rs/moq-video/src/decode/backend/openh264.rs

@@ -5,6 +5,7 @@
 //! inline ahead of each keyframe) and returns packed I420.
 
 use bytes::Bytes;
+use moq_net::Timestamp;
 use openh264::OpenH264API;
 use openh264::decoder::{Decoder, DecoderConfig};
 use openh264::formats::YUVSource;
@@ -33,7 +34,7 @@ impl Openh264 {
 }
 
 impl Backend for Openh264 {
-	fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
+	fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
 		let decoded = self
 			.decoder
 			.decode(&access_unit)
@@ -64,7 +65,7 @@ impl Backend for Openh264 {
 		);
 		// openh264 is one-in one-out, so the input timestamp is the output's.
 		Ok(vec![Decoded {
-			timestamp_us,
+			timestamp,
 			frame: Frame::I420(frame),
 		}])
 	}

rs/moq-video/src/decode/backend/videotoolbox.rs

@@ -22,6 +22,7 @@ use std::ptr::{self, NonNull};
 
 use bytes::Bytes;
 use moq_mux::codec::annexb::NalIterator;
+use moq_net::Timestamp;
 use objc2_core_foundation::{CFDictionary, CFNumber, CFNumberType, CFRetained, CFString};
 use objc2_core_media::{
 	CMBlockBuffer, CMFormatDescription, CMSampleBuffer, CMTime, CMVideoFormatDescriptionCreateFromH264ParameterSets,
@@ -166,7 +167,7 @@ impl VideoToolbox {
 }
 
 impl Backend for VideoToolbox {
-	fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
+	fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result<Vec<Decoded>, Error> {
 		// Split the Annex-B access unit, pull out any parameter sets, and gather
 		// the slices into length-prefixed (4-byte) form. `NalIterator` yields the
 		// parameter-set NALs as zero-copy `Bytes` (sub-slices of `access_unit`), so
@@ -230,7 +231,7 @@ impl Backend for VideoToolbox {
 		Ok(std::mem::take(&mut self.sink.frames)
 			.into_iter()
 			.map(|i420| Decoded {
-				timestamp_us,
+				timestamp,
 				frame: Frame::I420(i420),
 			})
 			.collect())

rs/moq-video/src/decode/consumer.rs

@@ -61,15 +61,10 @@ impl Consumer {
 			let Some(mux_frame) = self.track.read().await? else {
 				return Ok(None);
 			};
-			let timestamp_us: u64 = mux_frame
-				.timestamp
-				.as_micros()
-				.try_into()
-				.map_err(|_| moq_net::TimeOverflow)?;
 
 			self.pending.extend(
 				self.decoder
-					.decode(&mux_frame.payload, timestamp_us, mux_frame.keyframe)?,
+					.decode(&mux_frame.payload, mux_frame.timestamp, mux_frame.keyframe)?,
 			);
 		}
 	}

rs/moq-video/src/decode/decoder.rs

@@ -12,6 +12,7 @@ use std::time::Duration;
 use bytes::Bytes;
 use hang::catalog::{VideoCodec, VideoConfig};
 use moq_mux::codec::{annexb, h264, h265};
+use moq_net::Timestamp;
 
 use super::Frame;
 use super::backend::{self, Backend, Codec};
@@ -141,9 +142,12 @@ impl Decoder {
 		self.backend.name()
 	}
 
-	/// Decode one container frame, returning zero or more raw frames stamped with
-	/// `timestamp_us` (the container frame's presentation time in microseconds).
-	pub fn decode(&mut self, payload: &Bytes, timestamp_us: u64, keyframe: bool) -> Result<Vec<Frame>, Error> {
+	/// Decode one container frame, returning zero or more raw frames. `timestamp` is
+	/// this frame's presentation time; it rides through the decoder and comes back on
+	/// each output frame, so a reordering decoder (B-frames) stamps every picture
+	/// with its own presentation time rather than this access unit's. With no
+	/// reordering the two coincide.
+	pub fn decode(&mut self, payload: &Bytes, timestamp: Timestamp, keyframe: bool) -> Result<Vec<Frame>, Error> {
 		// Wait for the first keyframe: a decoder started mid-GOP can't decode
 		// delta frames, and the parameter sets ride along with the keyframe.
 		if !self.got_keyframe {
@@ -167,10 +171,10 @@ impl Decoder {
 
 		Ok(self
 			.backend
-			.decode(annexb, timestamp_us, keyframe)?
+			.decode(annexb, timestamp, keyframe)?
 			.into_iter()
 			.map(|decoded| Frame {
-				timestamp_us: decoded.timestamp_us,
+				timestamp: decoded.timestamp,
 				width: decoded.frame.width(),
 				height: decoded.frame.height(),
 				inner: decoded.frame,
@@ -181,6 +185,8 @@ impl Decoder {
 
 #[cfg(test)]
 mod tests {
+	use moq_net::Timestamp;
+
 	use super::backend::{self, Codec};
 	use crate::encode::{Config as EncodeConfig, Encoder, Kind as EncodeKind};
 	use crate::frame::I420;
@@ -220,15 +226,30 @@ mod tests {
 		let mut decoded = Vec::new();
 		for i in 0..10u64 {
 			let keyframe = i == 0;
+			// Distinct, spread-apart timestamps so a round-tripped value is unambiguous.
+			let timestamp = Timestamp::from_micros(i * 33_333).unwrap();
 			for packet in encoder.encode_rgba(&frame, 320, 240, keyframe).unwrap() {
-				decoded.extend(decoder.decode(packet, i * 33_333, keyframe).unwrap());
+				decoded.extend(decoder.decode(packet, timestamp, keyframe).unwrap());
 			}
 		}
 
 		assert!(!decoded.is_empty(), "decoder produced no frames");
 		for out in &decoded {
 			assert_gray(&out.frame.to_i420().unwrap(), 320, 240);
 		}
+
+		// The timestamp rides through the codec and comes back on each picture. These
+		// backends don't reorder, so it returns in feed order: strictly increasing and
+		// drawn from the values we fed.
+		let micros: Vec<u128> = decoded.iter().map(|d| d.timestamp.as_micros()).collect();
+		assert!(
+			micros.windows(2).all(|w| w[0] < w[1]),
+			"decoded timestamps not strictly increasing: {micros:?}"
+		);
+		assert!(
+			micros.iter().all(|&t| t % 33_333 == 0 && t < 333_330),
+			"decoded timestamp outside the fed set: {micros:?}"
+		);
 	}
 
 	/// A decoder config selecting one backend by kind.