feat(moq-net): mock up typed timed/untimed tracks

rs/hang/src/container/frame.rs

@@ -174,7 +174,7 @@ mod test {
 
 	#[test]
 	fn track_info_uses_container_timescale() {
-		assert_eq!(track_info(crate::catalog::PRIORITY.video).timescale, TIMESCALE);
+		assert_eq!(track_info(crate::catalog::PRIORITY.video).timescale, Some(TIMESCALE));
 	}
 
 	#[test]
@@ -186,7 +186,7 @@ mod test {
 		// Retimescaling for a container that carries the source's own scale keeps it, since that
 		// is the shape that would otherwise reach for `Info::default()` and lose the retention.
 		let at = track_info(crate::catalog::PRIORITY.video).with_timescale(Timescale::MILLI);
-		assert_eq!(at.timescale, Timescale::MILLI);
+		assert_eq!(at.timescale, Some(Timescale::MILLI));
 		assert_eq!(at.max_age, Some(MAX_AGE));
 	}
 

rs/moq-ffi/src/producer.rs

@@ -65,7 +65,7 @@ impl TryFrom<&moq_net::track::Info> for MoqTrackInfo {
 		Ok(Self {
 			priority: info.priority,
 			max_age_us,
-			timescale: Some(info.timescale.as_u64()),
+			timescale: info.timescale.map(|scale| scale.as_u64()),
 		})
 	}
 }

rs/moq-flate/src/stream/mod.rs

@@ -190,7 +190,7 @@ mod test {
 	/// allocating 32 MB to provoke it. Borrowed from moq-json's stream tests.
 	fn rejecting_track() -> moq_net::track::Producer {
 		let mut info = moq_net::track::Info::default();
-		info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap();
+		info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap());
 
 		moq_net::broadcast::Info::new()
 			.produce()

rs/moq-gst/src/sink/pad.rs

@@ -1248,7 +1248,7 @@ mod tests {
 			.subscribe(None)
 			.await
 			.expect("subscribe to the opaque track");
-		assert_eq!(subscriber.info().timescale, moq_net::Timescale::MICRO);
+		assert_eq!(subscriber.info().timescale, Some(moq_net::Timescale::MICRO));
 		assert_eq!(
 			subscriber.info().max_age,
 			Some(std::time::Duration::from_secs(5)),

rs/moq-json/src/snapshot/mod.rs

@@ -704,7 +704,7 @@ mod test {
 	/// allocating 32 MB to provoke it.
 	fn rejecting_track() -> moq_net::track::Producer {
 		let mut info = moq_net::track::Info::default();
-		info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap();
+		info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap());
 
 		moq_net::broadcast::Info::new()
 			.produce()

rs/moq-json/src/stream/mod.rs

@@ -237,7 +237,7 @@ mod test {
 	/// allocating 32 MB to provoke it.
 	fn rejecting_track() -> moq_net::track::Producer {
 		let mut info = moq_net::track::Info::default();
-		info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap();
+		info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap());
 
 		moq_net::broadcast::Info::new()
 			.produce()

rs/moq-json/src/window/mod.rs

@@ -91,7 +91,7 @@ mod test {
 	/// A track whose timestamp conversion rejects every frame after its group is published.
 	fn rejecting_track() -> moq_net::track::Producer {
 		let mut info = moq_net::track::Info::default();
-		info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap();
+		info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap());
 
 		moq_net::broadcast::Info::new()
 			.produce()

rs/moq-mux/src/catalog/producer.rs

@@ -1132,11 +1132,11 @@ mod test {
 
 		let info = catalog.track_info(hang::catalog::PRIORITY.video);
 		assert_eq!(info.max_age, Some(std::time::Duration::from_secs(3)));
-		assert_eq!(info.timescale, hang::container::TIMESCALE);
+		assert_eq!(info.timescale, Some(hang::container::TIMESCALE));
 
 		let at = info.with_timescale(moq_net::Timescale::MILLI);
 		assert_eq!(at.max_age, Some(std::time::Duration::from_secs(3)));
-		assert_eq!(at.timescale, moq_net::Timescale::MILLI);
+		assert_eq!(at.timescale, Some(moq_net::Timescale::MILLI));
 
 		// Every handle mints under the same policy, whatever order it was taken in: the codec
 		// paths hold a reservation and the container paths hold a clone.

rs/moq-net/src/ietf/publisher.rs

@@ -671,7 +671,7 @@ where
 				None => track::Position::after_group(end.group),
 			});
 			let _ = track.update(subscription);
-			let timescale = msg.properties_wanted.then(|| track.info().timescale);
+			let timescale = track.info().timescale.filter(|_| msg.properties_wanted);
 
 			// Draft-20 replaced joining FETCH with subscription fills. Older drafts save
 			// the same boundary used by the subscription so the two streams never overlap.
@@ -716,7 +716,7 @@ where
 						// We serve the newest group first, matching moq-lite.
 						true => ietf::Properties {
 							max_cache_duration: track.info().max_age,
-							timescale: Some(track.info().timescale),
+							timescale: track.info().timescale,
 							priority: Some(super::priority::to_wire(track.info().priority)),
 							group_order: Some(GroupOrder::Descending),
 						},
@@ -3279,7 +3279,9 @@ mod serve_tests {
 	fn serve(version: Version) -> Serve {
 		let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce();
 		let broadcast = origin.publish("room", crate::origin::Route::default()).unwrap();
-		let track = broadcast.create_track("video", None).unwrap();
+		// Timed, so served objects carry a Timestamp.
+		let info = track::Info::default().with_timescale(crate::Timescale::MILLI);
+		let track = broadcast.create_track("video", info).unwrap();
 
 		let session = ScriptedSession::per_stream(vec![Vec::new()]);
 		let log = session.log.clone();

rs/moq-net/src/ietf/subscriber.rs

@@ -1996,8 +1996,9 @@ where
 			priority,
 			largest,
 		} = accepted;
+		// Untimed unless SUBSCRIBE_OK declared a TIMESCALE.
 		let info = track::Info::default()
-			.with_timescale(Timescale::MICRO)
+			.with_timescale(timescale)
 			.with_max_age(max_age)
 			.with_priority(super::priority::from_wire(priority.unwrap_or(128)));
 		let (mut track, dynamic) = match resumed {
@@ -3504,7 +3505,7 @@ where
 		// have set it. FETCH_OK carries the same Track Properties, so the retention window
 		// comes from it.
 		let info = track::Info::default()
-			.with_timescale(Timescale::MICRO)
+			.with_timescale(ok.properties.timescale)
 			.with_max_age(ok.properties.max_cache_duration);
 		// Joined fetches still count until they pick the accepted group up from the cache.
 		let joined = request.result.clone();

rs/moq-net/src/lite/publisher.rs

@@ -1202,7 +1202,7 @@ impl<S: crate::transport::poll::Session> Request<S> for TrackInfoServe {
 		writer.buffer(&lite::TrackInfo {
 			priority: info.priority,
 			max_age: info.max_age,
-			timescale: info.timescale,
+			timescale: info.scale(),
 		})?;
 		Poll::Ready(Ok(ControlFlow::Break(())))
 	}
@@ -1295,7 +1295,7 @@ impl<S: crate::transport::poll::Session> Request<S> for SubscribeServe<S> {
 					// "don't emit the prefix" (the frames still carry timestamps in the
 					// model, just not on this wire).
 					let timescale = if shared.version.has_track_stream() {
-						Some(track.info().timescale)
+						Some(track.info().scale())
 					} else {
 						None
 					};

rs/moq-net/src/lite/subscriber.rs

@@ -3876,7 +3876,7 @@ impl<S: crate::transport::poll::Session> kio::Task for TrackServeRun<S> {
 							// Lite05 carries per-frame timestamps on the wire at this scale;
 							// `Some` tells the ingest to decode them instead of stamping
 							// local receive time.
-							let timescale = Some(info.timescale);
+							let timescale = info.timescale;
 							self.state = TrackRunState::Serve(ServeLoop::new(&self.serve, request, info, timescale));
 						}
 						Err(err) => {

rs/moq-net/src/lite/track.rs

@@ -203,7 +203,7 @@ mod test {
 		let info = TrackInfo {
 			priority: info.priority,
 			max_age: info.max_age,
-			timescale: info.timescale,
+			timescale: info.scale(),
 		};
 		let mut buf = Vec::new();
 		info.encode(&mut buf, Version::Lite05).unwrap();

rs/moq-net/src/model/group.rs

@@ -26,6 +26,8 @@ use std::task::{Poll, ready};
 
 use crate::{Error, IntoBytes, Result, Timestamp};
 
+pub use super::timing::group::{Timed, Untimed};
+
 /// Maximum total size of frames in a group.
 ///
 /// A write that would exceed this aborts the group with [`Error::GroupTooLarge`].
@@ -461,9 +463,9 @@ impl Producer {
 		self.info
 	}
 
-	/// The parent track's timescale.
+	/// The scale this group's frame timestamps are stored at; milliseconds on an untimed track.
 	pub fn timescale(&self) -> Timescale {
-		self.track.timescale
+		self.track.scale()
 	}
 
 	/// Start the group at frame `index` rather than 0, so the first frame written lands
@@ -506,7 +508,7 @@ impl Producer {
 	/// a presentation time, pass [`Timestamp::now`] explicitly.
 	pub fn write_frame<B: IntoBytes>(&mut self, timestamp: Timestamp, data: B) -> Result<()> {
 		let timestamp = timestamp
-			.convert(self.track.timescale)
+			.convert(self.track.scale())
 			.map_err(|_| Error::TimestampMismatch)?;
 		let payload = data.into_bytes();
 		if payload.len() as u64 > MAX_CACHE_BYTES {
@@ -561,7 +563,7 @@ impl Producer {
 		for frame in frames.filled() {
 			frame
 				.timestamp
-				.convert(self.track.timescale)
+				.convert(self.track.scale())
 				.map_err(|_| Error::TimestampMismatch)?;
 			if frame.payload.len() as u64 > MAX_CACHE_BYTES {
 				return Err(Error::FrameTooLarge);
@@ -590,7 +592,7 @@ impl Producer {
 		for mut frame in frames.drain() {
 			frame.timestamp = frame
 				.timestamp
-				.convert(self.track.timescale)
+				.convert(self.track.scale())
 				.expect("timestamp scale checked above");
 			let size = frame.payload.len() as u64;
 			state.cache += size;
@@ -654,7 +656,7 @@ impl Producer {
 	fn open_frame(&mut self, frame: frame::Info, buf: &FrameBuf) -> Result<frame::Info> {
 		let timestamp = frame
 			.timestamp
-			.convert(self.track.timescale)
+			.convert(self.track.scale())
 			.map_err(|_| Error::TimestampMismatch)?;
 		if frame.size > MAX_CACHE_BYTES {
 			return Err(Error::FrameTooLarge);
@@ -1435,9 +1437,9 @@ impl Consumer {
 		self.cursor.state.poll_closed(waiter)
 	}
 
-	/// The parent track's timescale.
+	/// The scale this group's frame timestamps are stored at; milliseconds on an untimed track.
 	pub fn timescale(&self) -> Timescale {
-		self.track.timescale
+		self.track.scale()
 	}
 
 	/// The index of the next frame this consumer will return.

rs/moq-net/src/model/mod.rs

@@ -4,6 +4,7 @@ pub mod cache;
 pub mod frame;
 pub mod group;
 pub mod track;
+pub mod typed;
 
 // The origin + announce subsystem shares one implementation (a route table).
 // It stays in a single private module and is surfaced as two curated public
@@ -22,6 +23,7 @@ mod resume;
 mod subscription;
 mod time;
 mod timed;
+mod timing;
 mod weak_cache;
 
 #[cfg(test)]

rs/moq-net/src/model/timed.rs

@@ -1,59 +1,21 @@
-use bytes::Bytes;
-
 use crate::Timestamp;
 
-/// A value to publish, and optionally when it was captured.
+/// A value and the time it was captured, written to a timed track.
 ///
-/// `T` is the clock the time is on: a raw [`Timestamp`] by default, or whatever a higher layer
-/// maps onto one. Converts from a bare payload (bytes, or a `&V` to serialize), so a producer
-/// taking one still accepts the plain value, stamped when written.
-#[derive(Debug, Clone, Copy)]
-pub struct Timed<P, T = Timestamp> {
+/// An untimed track takes the bare value instead, so a stamp can be neither forgotten on a
+/// timed track nor invented on an untimed one.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Timed<P> {
 	/// The value to publish.
 	pub value: P,
 
-	/// When the value was captured. `None` stamps it when written.
-	pub at: Option<T>,
-}
-
-impl<P, T> Timed<P, T> {
-	/// Set when the value was captured.
-	pub fn at(mut self, at: T) -> Self {
-		self.at = Some(at);
-		self
-	}
-}
-
-impl<B: Into<Bytes>, T> From<B> for Timed<Bytes, T> {
-	fn from(value: B) -> Self {
-		Self {
-			value: value.into(),
-			at: None,
-		}
-	}
+	/// When the value was captured, at any scale; the track converts it to its own.
+	pub at: Timestamp,
 }
 
-impl<'a, V, T> From<&'a V> for Timed<&'a V, T> {
-	fn from(value: &'a V) -> Self {
-		Self { value, at: None }
-	}
-}
-
-#[cfg(test)]
-mod tests {
-	use super::*;
-
-	#[test]
-	fn a_bare_payload_converts_untimed() {
-		let timed: Timed<Bytes> = vec![1u8, 2].into();
-		assert_eq!(timed.value, Bytes::from_static(&[1, 2]));
-		assert_eq!(timed.at, None);
-
-		let stamped = Timed::from(&b"x"[..]).at(Timestamp::from_millis(5).unwrap());
-		assert_eq!(stamped.at, Some(Timestamp::from_millis(5).unwrap()));
-
-		let value = 7u32;
-		let timed: Timed<&u32, std::time::Instant> = (&value).into();
-		assert_eq!(*timed.value, 7);
+impl<P> Timed<P> {
+	/// Pair a value with when it was captured.
+	pub fn new(value: P, at: Timestamp) -> Self {
+		Self { value, at }
 	}
 }