moq-net: serve one subscription and N fetches concurrently per track (#1634)

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

Author: kixelated

rs/moq-native/tests/broadcast.rs

@@ -300,7 +300,7 @@ async fn lite05_fetch_roundtrip(scheme: &str) {
 	// Fetch group 0 directly, without subscribing. No live producer holds the group
 	// on the client, so this issues a wire FETCH upstream.
 	let mut group_sub = tokio::time::timeout(TIMEOUT, async {
-		bc.track("video").unwrap().fetch(0, None).unwrap().await
+		bc.track("video").unwrap().fetch_group(0, None).unwrap().await
 	})
 	.await
 	.expect("fetch timed out")
@@ -346,6 +346,131 @@ async fn broadcast_moq_lite_05_fetch_webtransport() {
 	lite05_fetch_roundtrip("https").await;
 }
 
+/// A fetch must be served while a live subscription is active on the same track.
+/// The relay subscribes starting at the latest group, so an older group isn't
+/// cached and the fetch has to issue a wire FETCH concurrently with the
+/// subscription. Older relays served a subscription OR a fetch, never both, so
+/// this fetch would have hung.
+async fn lite05_fetch_during_subscribe(scheme: &str) {
+	use moq_native::moq_net::{Timescale, Timestamp};
+
+	fn timestamped_frame(us: u64, payload: &str) -> moq_net::Frame {
+		moq_net::Frame {
+			size: payload.len() as u64,
+			timestamp: Some(Timestamp::new(us, Timescale::MICRO).unwrap()),
+		}
+	}
+
+	let pub_origin = Origin::random().produce();
+	let mut broadcast = pub_origin.create_broadcast("test").expect("failed to create broadcast");
+	let mut track = broadcast
+		.create_track("video", moq_net::TrackInfo::default().with_timescale(Timescale::MICRO))
+		.expect("failed to create track");
+
+	// Group 0 is the "past" group only reachable via FETCH; group 1 is the latest,
+	// delivered live over the subscription.
+	let mut group0 = track.append_group().expect("append group 0"); // seq 0
+	let mut w = group0.create_frame(timestamped_frame(10_000, "old")).expect("frame 0");
+	w.write(bytes::Bytes::from_static(b"old")).expect("write 0");
+	w.finish().expect("finish frame 0");
+	group0.finish().expect("finish group 0");
+
+	let mut group1 = track.append_group().expect("append group 1"); // seq 1
+	let mut w = group1.create_frame(timestamped_frame(20_000, "new")).expect("frame 1");
+	w.write(bytes::Bytes::from_static(b"new")).expect("write 1");
+	w.finish().expect("finish frame 1");
+	group1.finish().expect("finish group 1");
+
+	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("failed to init server");
+	let addr = server.local_addr().expect("failed to get 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("failed to 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");
+
+	// Subscribe (starts at the latest group) and read the live group, which
+	// establishes the upstream subscription and leaves it active.
+	let mut track_sub = tokio::time::timeout(TIMEOUT, async {
+		bc.track("video").unwrap().subscribe(None).unwrap().await
+	})
+	.await
+	.expect("subscribe timed out")
+	.expect("subscribe failed");
+	let mut live = tokio::time::timeout(TIMEOUT, track_sub.recv_group())
+		.await
+		.expect("recv_group timed out")
+		.expect("recv_group failed")
+		.expect("track closed prematurely");
+	assert_eq!(live.sequence, 1);
+	let frame = tokio::time::timeout(TIMEOUT, live.read_frame())
+		.await
+		.expect("read_frame timed out")
+		.expect("read_frame failed")
+		.expect("group closed prematurely");
+	assert_eq!(&*frame, b"new");
+
+	// While the subscription is still held and active, fetch the older group. The
+	// relay doesn't have it cached (subscription started at the latest), so this
+	// must issue a wire FETCH concurrently with the live subscription.
+	let mut fetched = tokio::time::timeout(TIMEOUT, async {
+		bc.track("video").unwrap().fetch_group(0, None).unwrap().await
+	})
+	.await
+	.expect("fetch timed out")
+	.expect("fetch failed");
+	assert_eq!(fetched.sequence, 0);
+	let frame = tokio::time::timeout(TIMEOUT, fetched.read_frame())
+		.await
+		.expect("fetch read_frame timed out")
+		.expect("fetch read_frame failed")
+		.expect("fetched group closed prematurely");
+	assert_eq!(&*frame, b"old");
+
+	// The live subscription is unaffected: a freshly published group still arrives.
+	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_fetch_during_subscribe_webtransport() {
+	lite05_fetch_during_subscribe("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/lite/publisher.rs

@@ -582,7 +582,7 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 
 		let group = broadcast
 			.track(&fetch.track)?
-			.fetch(
+			.fetch_group(
 				fetch.group,
 				crate::Fetch {
 					priority: fetch.priority,