fix(transcode): refuse a fetch that starts mid-group (backport #4812)

rs/moq-transcode/src/lib.rs

@@ -1271,6 +1271,129 @@ mod tests {
 		transcoder.abort();
 	}
 
+	/// A fetch that starts partway through a group is refused rather than served.
+	/// A fresh encode of that group need not match the bytes of the head a reader
+	/// holds, so its tail cannot continue it. The reader moves on to the next group.
+	#[tokio::test]
+	async fn a_mid_group_fetch_is_refused() {
+		let source = source_catalog(320, 240);
+		// Serve any source group on demand, so a fetch that is not refused has
+		// something to transcode and resolves.
+		let source_fetches = source._track.dynamic();
+		let server = tokio::spawn(async move {
+			while let Ok(request) = source_fetches.requested_group().await {
+				let mut group = request.accept(None).unwrap();
+				write_keyframe(&mut group);
+				group.finish().unwrap();
+			}
+		});
+
+		let config = Config {
+			ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(),
+			encoder: moq_video::encode::Kind::Software,
+			decoder: moq_video::decode::Kind::Software,
+			source: None,
+			..Default::default()
+		};
+		let output = moq_net::broadcast::Info::default().produce();
+		let consumer = output.consume();
+		let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config));
+
+		let catalog = loop {
+			match consumer.track(hang::Catalog::DEFAULT_NAME) {
+				Ok(track) => break track,
+				Err(moq_net::Error::NotFound) => tokio::task::yield_now().await,
+				Err(err) => panic!("catalog track: {err}"),
+			}
+		};
+		let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(catalog.subscribe(None).await.unwrap());
+		await_catalog(&mut catalogs, |snapshot| {
+			snapshot.video.renditions.contains_key("video/120p")
+		})
+		.await;
+
+		// The source publishes no live groups, so group 7 is a cache miss that
+		// reaches the rung's fetch handler.
+		let rung = consumer.track("video/120p").unwrap();
+		rung.query().await.unwrap();
+
+		let partial = rung
+			.fetch_group(7, moq_net::group::Fetch::default().with_frame_start(2))
+			.await;
+		match partial {
+			Err(moq_net::Error::NotFound) => {}
+			Err(err) => panic!("expected a NotFound refusal, got {err}"),
+			Ok(_) => panic!("served a fetch that starts mid-group"),
+		}
+
+		// The whole group still serves.
+		let mut whole = rung.fetch_group(7, None).await.unwrap();
+		while whole.read_frame().await.unwrap().is_some() {}
+		assert!(whole.finished().await.unwrap() > 0, "the whole group had no frames");
+
+		server.abort();
+		transcoder.abort();
+	}
+
+	/// Two transcoders fed the same source publish groups that mirror the
+	/// source's sequences and timestamps rather than anything numbered per
+	/// instance.
+	#[tokio::test]
+	async fn two_instances_mirror_source_groups() {
+		let source = source_broadcast(2, 5);
+		let config = Config {
+			ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(),
+			encoder: moq_video::encode::Kind::Software,
+			decoder: moq_video::decode::Kind::Software,
+			source: None,
+			..Default::default()
+		};
+
+		// Mirrored from the source, not numbered by either instance.
+		let expected: Vec<_> = (0..2u64)
+			.map(|sequence| {
+				(
+					sequence,
+					(0..5).map(|i| (sequence * 5 + i) as u128 * 33_333).collect::<Vec<_>>(),
+				)
+			})
+			.collect();
+
+		for _ in 0..2 {
+			let output = moq_net::broadcast::Info::default().produce();
+			let consumer = output.consume();
+			let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config.clone()));
+
+			let track = loop {
+				match consumer.track(hang::Catalog::DEFAULT_NAME) {
+					Ok(track) => break track,
+					Err(moq_net::Error::NotFound) => tokio::task::yield_now().await,
+					Err(err) => panic!("catalog track: {err}"),
+				}
+			};
+			let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(track.subscribe(None).await.unwrap());
+			await_catalog(&mut catalogs, |snapshot| {
+				snapshot.video.renditions.contains_key("video/120p")
+			})
+			.await;
+
+			let rung = consumer.track("video/120p").unwrap();
+			let mut groups = Vec::new();
+			for sequence in 0..2 {
+				let mut fetched = rung.fetch_group(sequence, None).await.unwrap();
+				let mut timestamps = Vec::new();
+				while let Some(payload) = fetched.read_frame().await.unwrap() {
+					let frame = hang::container::Frame::decode(payload.payload).unwrap();
+					timestamps.push(frame.timestamp.as_micros());
+				}
+				groups.push((fetched.sequence, timestamps));
+			}
+
+			assert_eq!(groups, expected);
+			transcoder.abort();
+		}
+	}
+
 	/// A source whose codec description changes rebuilds the shared decode, so
 	/// every rung retires with it. The picture may not have moved at all, so shape
 	/// alone would hand the replacements the names that just ended. They have to be

rs/moq-transcode/src/rung.rs

@@ -7,7 +7,9 @@
 //!   source track (mirroring the aggregate subscription) and transcodes group
 //!   for group until the track goes `unused` again.
 //! - A fetch of a specific group (`requested_group`) fetches that same group
-//!   from the source and transcodes just that group with a fresh encoder.
+//!   from the source and transcodes just that group with a fresh encoder. A
+//!   fetch that starts mid-group is refused: a fresh encode's frames are only
+//!   valid after the head that same encode produced.
 //!
 //! Output groups mirror the source group sequence numbers 1:1, so a fetch for
 //! output group N maps to source group N and a player switching renditions
@@ -481,6 +483,20 @@ fn spawn_fetch(
 /// "the handler vanished" and hides the actual decode/encode/source failure from
 /// the waiting consumer.
 async fn fetch(rung: Rung, request: moq_net::group::Request) -> Result<(), Error> {
+	// A fresh encode of this group need not match the bytes the reader's head
+	// came from (the live encoder, or an earlier fetch), so its tail cannot
+	// continue that head. Refuse it so the reader moves on to the next group
+	// boundary instead of decoding a splice.
+	if request.frame_start() != 0 {
+		tracing::debug!(
+			sequence = request.sequence(),
+			frame = request.frame_start(),
+			"refusing a fetch that starts mid-group"
+		);
+		request.reject(moq_net::Error::NotFound);
+		return Ok(());
+	}
+
 	let options = moq_net::group::Fetch::default().with_priority(request.priority());
 	let mut source = match rung.source.fetch_group(request.sequence(), options).await {
 		Ok(source) => source,