From dd88fa3b144bac3010497d2aeda3671ffa21a613 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 16:17:41 -0700 Subject: [PATCH] fix(transcode): refuse a fetch that starts mid-group (#4812) Drops the quest/ file change, since release does not carry that quest. (cherry picked from commit 4ae871c947a58732cac9fd9d4d82b619695f639e) Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/lib.rs | 123 +++++++++++++++++++++++++++++++++++ rs/moq-transcode/src/rung.rs | 18 ++++- 2 files changed, 140 insertions(+), 1 deletion(-) diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index 2fbbd371cb..9587f07b59 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/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::>(), + ) + }) + .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 diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 1fc217c823..4176e0eba6 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/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,