moq-lite-05: move immutable track props to a Track Stream (TRACK_INFO)

Replace the per-response publisher metadata in SUBSCRIBE_OK with a
dedicated, on-demand Track Stream, per moq-dev/drafts#25. Scoped to the
WIP Lite05 version; Lite01-04 keep SUBSCRIBE_OK unchanged.

- New Track Stream (0x6): a TRACK request (broadcast path + track name)
  answered with a single TRACK_INFO carrying the immutable publisher
  properties (Priority, Ordered, Cache, Timescale, Compression), then a
  FIN (or reset on error / missing track).
- Removed the static props (compression/timescale/cache) from SUBSCRIBE_OK;
  on Lite05 a subscription is accepted implicitly (rejection is a reset)
  and the publisher sends nothing on the subscribe stream.
- Subscriber flights TRACK and SUBSCRIBE in parallel, so the first group
  still arrives in one round trip. A pending TrackEntry is inserted before
  SUBSCRIBE, so group streams that race ahead of TRACK_INFO park on a
  resolved channel (buffered by QUIC flow control) instead of being
  dropped. The resolved (producer, compression, timescale) is reused for
  every group's decode instead of being re-derived per response, and is
  fetched once for the upstream subscription's lifetime (linger).

Publisher resolves TRACK_INFO by subscribing to read the track's .info and
dropping the subscription; a parallel SUBSCRIBE coalesces onto the same
upstream producer. Priority/Ordered are sent as 0/false for now since the
model Track carries no publisher priority/order field yet.

Not included (no functional gap in the Rust impl, which never resolves a
group range or emits drops): SUBSCRIBE_START/SUBSCRIBE_END and the
SUBSCRIBE_DROP renumber. Cross-package sync to js/net and doc/concept is
also deferred.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: kixelated

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

@@ -19,6 +19,7 @@ mod session;
 mod stream;
 mod subscribe;
 mod subscriber;
+mod track;
 mod version;
 
 pub use announce::*;
@@ -37,4 +38,6 @@ pub(super) use session::*;
 pub use stream::*;
 pub use subscribe::*;
 use subscriber::*;
+#[allow(unused_imports)]
+pub use track::*;
 pub use version::Version;

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

@@ -62,6 +62,7 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 			if let Err(err) = match kind {
 				lite::ControlType::Announce => self.recv_announce(stream).await,
 				lite::ControlType::Subscribe => self.recv_subscribe(stream).await,
+				lite::ControlType::Track => self.recv_track(stream).await,
 				lite::ControlType::Probe => {
 					self.recv_probe(stream);
 					Ok(())
@@ -455,8 +456,9 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 			.await?;
 
 		// Compress only when the producer marked the track worth it and the
-		// negotiated draft understands the SUBSCRIBE_OK codec field. Older drafts
-		// (lite-04 and below) get None and the frames stream verbatim.
+		// negotiated draft understands the codec field (carried in SUBSCRIBE_OK on
+		// lite-04 and below, in TRACK_INFO on lite-05+). Older drafts without it
+		// get None and the frames stream verbatim.
 		let supports_compression = !matches!(
 			version,
 			Version::Lite01 | Version::Lite02 | Version::Lite03 | Version::Lite04
@@ -477,20 +479,23 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 			None
 		};
 
-		let info = lite::SubscribeOk {
-			priority: subscribe.priority,
-			ordered: false,
-			max_latency: std::time::Duration::ZERO,
-			start_group: None,
-			end_group: None,
-			compression,
-			timescale,
-			// Announce the publisher's cache window so the subscriber (a relay)
-			// re-serves with the same eviction window. Pre-lite-05 peers ignore it.
-			cache: track.cache,
-		};
+		// Lite05+ accepts a subscription implicitly (rejection is a stream reset)
+		// and serves the immutable properties over a TRACK_INFO stream instead.
+		// Older drafts confirm acceptance with SUBSCRIBE_OK here.
+		if matches!(
+			version,
+			Version::Lite01 | Version::Lite02 | Version::Lite03 | Version::Lite04
+		) {
+			let info = lite::SubscribeOk {
+				priority: subscribe.priority,
+				ordered: false,
+				max_latency: std::time::Duration::ZERO,
+				start_group: None,
+				end_group: None,
+			};
 
-		stream.writer.encode(&lite::SubscribeResponse::Ok(info)).await?;
+			stream.writer.encode(&lite::SubscribeResponse::Ok(info)).await?;
+		}
 
 		// Track-level subscriber priority. SUBSCRIBE_UPDATE messages broadcast new values
 		// to both run_track (so future groups inherit the new priority) and serve_group
@@ -528,6 +533,84 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 		stream.writer.finish()?;
 		stream.writer.closed().await
 	}
+
+	/// Serve a Track Stream: reply with the track's immutable [`lite::TrackInfo`]
+	/// and FIN, or reset on error (e.g. the track does not exist). Lite05+ only.
+	pub async fn recv_track(&mut self, mut stream: Stream<S, Version>) -> Result<(), Error> {
+		let req = stream.reader.decode::<lite::Track>().await?;
+
+		let track = req.track.to_string();
+		let absolute = self.origin.absolute(&req.broadcast).to_owned();
+
+		tracing::debug!(broadcast = %absolute, %track, "track info requested");
+
+		let broadcast = self.origin.get_broadcast(&req.broadcast);
+		let version = self.version;
+
+		web_async::spawn(async move {
+			if let Err(err) = Self::run_track_info(&mut stream, &track, broadcast, version).await {
+				match &err {
+					Error::Cancel | Error::Transport(_) => {
+						tracing::debug!(broadcast = %absolute, %track, "track info cancelled")
+					}
+					err => tracing::warn!(broadcast = %absolute, %track, %err, "track info error"),
+				}
+				stream.writer.abort(&err);
+			}
+		});
+
+		Ok(())
+	}
+
+	async fn run_track_info(
+		stream: &mut Stream<S, Version>,
+		track_name: &str,
+		consumer: Option<BroadcastConsumer>,
+		version: Version,
+	) -> Result<(), Error> {
+		let broadcast = consumer.ok_or(Error::NotFound)?;
+
+		// The immutable properties are delivered when the (possibly dynamic)
+		// producer is accepted, so resolving them means subscribing. We read them
+		// off the subscriber and drop it immediately; a SUBSCRIBE flighted in
+		// parallel coalesces onto the same upstream producer, which linger keeps
+		// alive across this brief gap.
+		let track = broadcast
+			.consume_track(track_name)
+			.subscribe(crate::Subscription::default())
+			.await?;
+
+		// Mirror the negotiation in `run_subscribe` so the subscriber decodes
+		// frames the same way it'll see them served.
+		let supports_compression = !matches!(
+			version,
+			Version::Lite01 | Version::Lite02 | Version::Lite03 | Version::Lite04
+		);
+		let compression = if track.compress && supports_compression {
+			Compression::Deflate
+		} else {
+			Compression::None
+		};
+		let timescale = if version.has_timestamps() {
+			track.timescale
+		} else {
+			None
+		};
+
+		let info = lite::TrackInfo {
+			// The model carries no publisher-chosen priority/order yet, so both
+			// default to the tie-break-neutral values.
+			priority: 0,
+			ordered: false,
+			cache: track.cache,
+			timescale,
+			compression,
+		};
+
+		stream.writer.encode(&info).await?;
+		stream.writer.finish()?;
+		stream.writer.closed().await
+	}
 }
 
 /// Shared per-subscription state for the publisher side. Cloned (cheaply — every

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

@@ -13,6 +13,9 @@ pub enum ControlType {
 	Fetch = 3,
 	Probe = 4,
 	Goaway = 5,
+	/// Track Stream: a subscriber requests a track's immutable publisher
+	/// properties (TRACK_INFO) without subscribing or fetching. Lite05+ only.
+	Track = 6,
 }
 
 impl Decode<Version> for ControlType {

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

@@ -1,7 +1,7 @@
 use std::borrow::Cow;
 
 use crate::{
-	Compression, Path, Timescale,
+	Path,
 	coding::{Decode, DecodeError, Encode, EncodeError, Sizer},
 };
 
@@ -72,28 +72,18 @@ impl Message for Subscribe<'_> {
 	}
 }
 
+/// Sent by the publisher to accept a subscription (Lite01-04 only).
+///
+/// On Lite05+ a subscription is accepted implicitly (rejection is a stream
+/// reset) and the immutable publisher properties moved to [`TrackInfo`], fetched
+/// once over a [Track Stream](super::Track). This message is no longer sent.
 #[derive(Clone, Debug)]
 pub struct SubscribeOk {
 	pub priority: u8,
 	pub ordered: bool,
 	pub max_latency: std::time::Duration,
 	pub start_group: Option<u64>,
 	pub end_group: Option<u64>,
-	/// Codec the publisher will use for every frame on this track. Negotiated
-	/// here (not in SUBSCRIBE) so the subscriber blocks on this message before it
-	/// can decode any frame payload. Lite05+ only; older drafts always get
-	/// [`Compression::None`].
-	pub compression: Compression,
-	/// Per-frame timestamp scale advertised by the publisher. `None` means the
-	/// publisher doesn't carry per-frame timestamps on the wire (so frame
-	/// headers omit them). Lite05+ only; older drafts always decode as `None`.
-	/// On the wire `None` is `0` and `Some(n)` is `n`.
-	pub timescale: Option<Timescale>,
-	/// How long the publisher keeps old groups available before evicting them.
-	/// A relay re-serves with the same window and clamps each subscriber's stale
-	/// preference to it. Lite05+ only; older drafts always get
-	/// [`crate::DEFAULT_CACHE`].
-	pub cache: std::time::Duration,
 }
 
 impl Message for SubscribeOk {
@@ -103,23 +93,14 @@ impl Message for SubscribeOk {
 				self.priority.encode(w, version)?;
 			}
 			Version::Lite02 => {}
-			Version::Lite03 | Version::Lite04 => {
-				self.priority.encode(w, version)?;
-				(self.ordered as u8).encode(w, version)?;
-				self.max_latency.encode(w, version)?;
-				self.start_group.encode(w, version)?;
-				self.end_group.encode(w, version)?;
-			}
+			// Lite05+ no longer sends SUBSCRIBE_OK, but keep the Lite03/04 layout as
+			// the forward default so an accidental encode stays well-formed.
 			_ => {
 				self.priority.encode(w, version)?;
 				(self.ordered as u8).encode(w, version)?;
 				self.max_latency.encode(w, version)?;
 				self.start_group.encode(w, version)?;
 				self.end_group.encode(w, version)?;
-				// Order matches draft-lcurley-moq-lite-05 SUBSCRIBE_OK: Timescale, Cache, Compression.
-				self.timescale.map(u64::from).unwrap_or(0).encode(w, version)?;
-				self.cache.encode(w, version)?;
-				self.compression.to_code().encode(w, version)?;
 			}
 		}
 
@@ -134,61 +115,21 @@ impl Message for SubscribeOk {
 				max_latency: std::time::Duration::ZERO,
 				start_group: None,
 				end_group: None,
-				compression: Compression::None,
-				timescale: None,
-				cache: crate::DEFAULT_CACHE,
 			}),
 			Version::Lite02 => Ok(Self {
 				priority: 0,
 				ordered: false,
 				max_latency: std::time::Duration::ZERO,
 				start_group: None,
 				end_group: None,
-				compression: Compression::None,
-				timescale: None,
-				cache: crate::DEFAULT_CACHE,
 			}),
-			Version::Lite03 | Version::Lite04 => {
-				let priority = u8::decode(r, version)?;
-				let ordered = u8::decode(r, version)? != 0;
-				let max_latency = std::time::Duration::decode(r, version)?;
-				let start_group = Option::<u64>::decode(r, version)?;
-				let end_group = Option::<u64>::decode(r, version)?;
-
-				Ok(Self {
-					priority,
-					ordered,
-					max_latency,
-					start_group,
-					end_group,
-					compression: Compression::None,
-					timescale: None,
-					cache: crate::DEFAULT_CACHE,
-				})
-			}
-			_ => {
-				let priority = u8::decode(r, version)?;
-				let ordered = u8::decode(r, version)? != 0;
-				let max_latency = std::time::Duration::decode(r, version)?;
-				let start_group = Option::<u64>::decode(r, version)?;
-				let end_group = Option::<u64>::decode(r, version)?;
-				// Order matches draft-lcurley-moq-lite-05 SUBSCRIBE_OK: Timescale, Cache, Compression.
-				let timescale = Timescale::new(u64::decode(r, version)?).ok();
-				let cache = std::time::Duration::decode(r, version)?;
-				let compression =
-					Compression::from_code(u64::decode(r, version)?).map_err(|_| DecodeError::InvalidValue)?;
-
-				Ok(Self {
-					priority,
-					ordered,
-					max_latency,
-					start_group,
-					end_group,
-					compression,
-					timescale,
-					cache,
-				})
-			}
+			_ => Ok(Self {
+				priority: u8::decode(r, version)?,
+				ordered: u8::decode(r, version)? != 0,
+				max_latency: std::time::Duration::decode(r, version)?,
+				start_group: Option::<u64>::decode(r, version)?,
+				end_group: Option::<u64>::decode(r, version)?,
+			}),
 		}
 	}
 }
@@ -398,9 +339,6 @@ mod test {
 			max_latency: Duration::from_millis(250),
 			start_group: Some(3),
 			end_group: None,
-			compression: Compression::Deflate,
-			timescale: Some(Timescale::MICRO),
-			cache: Duration::from_secs(10),
 		}
 	}
 
@@ -412,60 +350,12 @@ mod test {
 	}
 
 	#[test]
-	fn compression_roundtrips_on_lite05() {
-		let got = roundtrip(Version::Lite05Wip, &sample());
-		assert_eq!(got.compression, Compression::Deflate);
+	fn fields_roundtrip_on_lite04() {
+		let got = roundtrip(Version::Lite04, &sample());
 		assert_eq!(got.priority, 7);
 		assert!(got.ordered);
+		assert_eq!(got.max_latency, Duration::from_millis(250));
 		assert_eq!(got.start_group, Some(3));
 		assert_eq!(got.end_group, None);
 	}
-
-	#[test]
-	fn compression_absent_before_lite05() {
-		let ok = sample();
-
-		// The compression varint only exists on lite-05+, so the older encoding is
-		// strictly shorter and always decodes back as None.
-		let mut buf04 = Vec::new();
-		ok.encode_msg(&mut buf04, Version::Lite04).unwrap();
-		let mut buf05 = Vec::new();
-		ok.encode_msg(&mut buf05, Version::Lite05Wip).unwrap();
-		assert!(
-			buf05.len() > buf04.len(),
-			"lite-05 carries extra compression + timescale varints"
-		);
-
-		assert_eq!(roundtrip(Version::Lite04, &ok).compression, Compression::None);
-	}
-
-	#[test]
-	fn timescale_roundtrips_on_lite05() {
-		let got = roundtrip(Version::Lite05Wip, &sample());
-		assert_eq!(got.timescale, Some(Timescale::MICRO));
-	}
-
-	#[test]
-	fn timescale_absent_before_lite05() {
-		// Lite04 doesn't carry the timescale varint, so it always decodes as None.
-		assert_eq!(roundtrip(Version::Lite04, &sample()).timescale, None);
-	}
-
-	#[test]
-	fn timescale_zero_on_wire_decodes_as_none() {
-		let mut ok = sample();
-		ok.timescale = None;
-		assert_eq!(roundtrip(Version::Lite05Wip, &ok).timescale, None);
-	}
-
-	#[test]
-	fn cache_roundtrips_on_lite05() {
-		assert_eq!(roundtrip(Version::Lite05Wip, &sample()).cache, Duration::from_secs(10));
-	}
-
-	#[test]
-	fn cache_absent_before_lite05() {
-		// Lite04 doesn't carry the cache varint, so it always decodes as the default.
-		assert_eq!(roundtrip(Version::Lite04, &sample()).cache, crate::DEFAULT_CACHE);
-	}
 }