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

rs/moq-native/tests/broadcast.rs

@@ -227,6 +227,104 @@ async fn broadcast_moq_lite_05_timestamps_webtransport() {
 	lite05_timestamp_roundtrip("https").await;
 }
 
+/// Lite05 Track Stream: the client resolves a track's immutable properties via
+/// `info()` (a TRACK request answered with TRACK_INFO) without subscribing, then
+/// subscribes and reads a frame. Exercises the on-demand info path end-to-end.
+async fn lite05_info_roundtrip(scheme: &str) {
+	use moq_native::moq_net::Timescale;
+
+	let pub_origin = Origin::random().produce();
+	let mut broadcast = pub_origin.create_broadcast("test").expect("create broadcast");
+	let mut track = broadcast
+		.create_track(Track::new("video").with_timescale(Timescale::MICRO))
+		.expect("create track");
+
+	let mut group = track.append_group().expect("append group");
+	let frame = moq_native::moq_net::Frame {
+		size: 5,
+		timestamp: Some(moq_native::moq_net::Timestamp::new(10_000, Timescale::MICRO).unwrap()),
+	};
+	let mut writer = group.create_frame(frame).expect("create frame");
+	writer.write(bytes::Bytes::from_static(b"hello")).expect("write frame");
+	writer.finish().expect("finish frame");
+	group.finish().expect("finish group");
+
+	let mut server_config = moq_native::ServerConfig::default();
+	server_config.bind = Some("[::]:0".to_string());
+	server_config.tls.generate = vec!["localhost".into()];
+	server_config.version = vec!["moq-lite-05-wip".parse().unwrap()];
+	let mut server = server_config.init().expect("init server");
+	let addr = server.local_addr().expect("local addr");
+
+	let sub_origin = Origin::random().produce();
+	let mut announcements = sub_origin.consume().announced();
+
+	let mut client_config = moq_native::ClientConfig::default();
+	client_config.tls.disable_verify = Some(true);
+	client_config.version = vec!["moq-lite-05-wip".parse().unwrap()];
+	let client = client_config.init().expect("init client");
+	let url: url::Url = format!("{scheme}://localhost:{}", addr.port()).parse().unwrap();
+
+	let server_handle = tokio::spawn(async move {
+		let request = server.accept().await.expect("no incoming connection");
+		let session = request.with_publisher(pub_origin.clone()).ok().await?;
+		let _broadcast = broadcast;
+		let _track = track;
+		let _ = session.closed().await;
+		Ok::<_, anyhow::Error>(())
+	});
+
+	let client = client.with_consumer(sub_origin);
+	let session = tokio::time::timeout(TIMEOUT, client.connect(url))
+		.await
+		.expect("client connect timed out")
+		.expect("client connect failed");
+
+	let (path, bc) = tokio::time::timeout(TIMEOUT, announcements.next())
+		.await
+		.expect("announce timed out")
+		.expect("origin closed");
+	assert_eq!(path.as_str(), "test");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
+
+	// Resolve the track's immutable info without subscribing.
+	let info = tokio::time::timeout(TIMEOUT, bc.consume_track("video").info())
+		.await
+		.expect("info timed out")
+		.expect("info failed");
+	assert_eq!(info.timescale, Some(Timescale::MICRO));
+
+	// A subscribe still works (and reuses the now-cached info).
+	let mut track_sub = bc
+		.consume_track("video")
+		.subscribe(None)
+		.await
+		.expect("subscribe failed");
+	let mut group_sub = tokio::time::timeout(TIMEOUT, track_sub.recv_group())
+		.await
+		.expect("recv_group timed out")
+		.expect("recv_group failed")
+		.expect("track closed prematurely");
+	let frame = tokio::time::timeout(TIMEOUT, group_sub.read_frame())
+		.await
+		.expect("read_frame timed out")
+		.expect("read_frame failed")
+		.expect("group closed prematurely");
+	assert_eq!(&*frame, b"hello");
+
+	drop(session);
+	server_handle
+		.await
+		.expect("server task panicked")
+		.expect("server task failed");
+}
+
+#[tracing_test::traced_test]
+#[tokio::test]
+async fn broadcast_moq_lite_05_info_webtransport() {
+	lite05_info_roundtrip("https").await;
+}
+
 /// On Lite05 a publisher that doesn't advertise a timescale still works:
 /// SUBSCRIBE_OK carries `timescale = 0` and neither side encodes a
 /// per-frame timestamp byte. Subscribers receive `frame.timestamp = None`.

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

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use crate::{
 	Broadcast, BroadcastDynamic, Error, Frame, FrameProducer, Group, GroupProducer, MAX_FRAME_SIZE, OriginProducer,
-	Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, Track, TrackProducer, TrackRequest,
+	Path, PathOwned, PendingTrack, StatsHandle, SubscriberStats, SubscriberTrack, Track, TrackProducer,
 	coding::{Reader, Stream},
 	ietf::{self, Control, FilterType, GroupOrder, RequestId},
 	model::BroadcastProducer,
@@ -536,7 +536,7 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		Ok(())
 	}
 
-	async fn run_subscribe(&mut self, broadcast_path: Path<'_>, broadcast: BroadcastDynamic, request: TrackRequest) {
+	async fn run_subscribe(&mut self, broadcast_path: Path<'_>, broadcast: BroadcastDynamic, request: PendingTrack) {
 		// Accept right away: IETF group data can arrive before SubscribeOk, so we
 		// need the producer in place to route it. This also unblocks the
 		// downstream subscriber's `consume_track`.

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

@@ -80,6 +80,7 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 		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).await;
 				Ok(())
@@ -461,8 +462,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
@@ -487,20 +489,23 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 		// broadcast. Dropping this guard (subscription end) releases it.
 		let _broadcast_sub = broadcasts.subscribe(&absolute);
 
-		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
@@ -538,6 +543,80 @@ 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.
+	///
+	/// Runs inline in its own control-stream task (see [`Self::handle`]); resolving
+	/// the info can be a cold upstream TRACK fetch, but that only blocks this task.
+	pub async fn recv_track(&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);
+
+		if let Err(err) = Self::run_track_info(&mut stream, &track, broadcast, self.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)?;
+
+		// Resolve the immutable properties without subscribing. Warm (a producer
+		// exists or the info is cached) this returns with no round trip; cold (a
+		// relay with no prior subscription) it triggers a single upstream TRACK
+		// fetch via the model's info-request channel.
+		let track = broadcast.consume_track(track_name).info().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 {