From 8e2e0fb68075861d457e45d9cc42f8bd4ed38f7a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 19 Oct 2026 00:54:25 -0700 Subject: [PATCH 1/5] chore(quest): claim m1/transcode-open-group Co-Authored-By: Claude Opus 5.5 From 8fdd4167cd024f079abec55a0984998f65c17f19 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 19 Oct 2026 01:01:48 -0700 Subject: [PATCH 2/5] fix(transcode): finish the open live group when demand ends A fetch is demand for the rung's live loop, but the group consumer it resolves to is not. When the fetch ended mid-group the loop aborted the group it was writing, under a reader that had already picked it up from the cache, and a later fetch re-transcoded the sequence under the same name. The loop now rides the open group out, as it already does on retirement, and leaves its source at the group boundary; demand returning before then resumes the session. Drops the held query the mirroring test needed to keep the live path from aborting between fetches. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/transcode-open-group.md | 28 --------- rs/moq-transcode/src/lib.rs | 97 +++++++++++++++++++++++++++++++- rs/moq-transcode/src/rung.rs | 42 ++++++++++---- 4 files changed, 126 insertions(+), 42 deletions(-) delete mode 100644 quest/m1/transcode-open-group.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 0230ba6414c..50c95717b62 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -101,7 +101,6 @@ lands before the cut. - [GOAWAY coverage](/quest/m1/goaway-coverage.md) - every IETF request stream handles GOAWAY, a closed control stream is a violation, and the draft-18 cutoff is honored - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust - [Decode errors end the reader](/quest/m1/decode-errors.md) - a container decode error fails the reader in Rust and JS instead of reading as loss -- [Transcode finishes its group](/quest/m1/transcode-open-group.md) - `moq-transcode` never aborts a group a fetch's live loop opened - [Rust request credit](/quest/m1/rs-request-credit.md) - a moq-net request fails at once without stream credit instead of waiting - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome - [Bufferbloat harness](/quest/m1/bufferbloat.md) - acceptance and performance checks impair paths with a rate limit and a deep queue, not random loss diff --git a/quest/m1/transcode-open-group.md b/quest/m1/transcode-open-group.md deleted file mode 100644 index 44bd7e008a5..00000000000 --- a/quest/m1/transcode-open-group.md +++ /dev/null @@ -1,28 +0,0 @@ -# [S] moq-transcode finishes the group it is transcoding - -## Goal - -When demand for a transcoded rendition ends mid-group, `moq-transcode` -finishes the open output group before leaving its source session. A later -fetch of that sequence gets the complete group, never an aborted one. - -## Plan - -Since #5820 a fetch counts as demand for its whole lifetime, so a fetch -starts the rung's live loop. When the fetch ends, the loop sees the track -unused and aborts its open output group with `Error::Cancel`; a later fetch -of that sequence hits the aborted group. #5820's tests worked around it by -holding queries open. - -Decided 2026-10-17 (maintainer): finish the group, then leave. A group name -always means the same content, so the published group must be the whole -group for every later reader, not whatever was transcoded before demand -went away. Rejected: aborting and re-transcoding on the next fetch (two -contents under one name), and serving fetches on a separate path (more -machinery for the same contract). - -Drop the held-query workaround in the tests once the loop finishes groups; -a regression test fetches a sequence after an earlier fetch that started -the live loop has ended. - -Public API: none. Wire: none. diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index a7ff11564ab..6b71197d2f4 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -10,7 +10,8 @@ //! - Subscribing to a rung attaches it to a shared live decode of the source //! (one subscription and one decoder per source, no matter how many rungs //! are active); each rung resizes and encodes its own copy, group for group, -//! stopping when the last subscriber leaves. +//! stopping at the end of the group in progress once the last subscriber +//! leaves. //! - Fetching a specific group fetches that same group from the source and //! transcodes just that group, unless every caller leaves before the //! request is accepted, in which case nothing is encoded. Output groups @@ -1536,6 +1537,98 @@ mod tests { transcoder.abort(); } + /// A fetch is demand, so it starts the live path, which claims the group the + /// source has open. A reader that picked that group up from the cache is not + /// demand, so when the fetch ends the live path is left mid-group with + /// nobody asking. It finishes the group before leaving its source: a group + /// name always means the same content, so the reader gets the whole group + /// rather than one aborted under it. + /// + /// Paused so the runtime settles before the source group ends: the live path + /// has seen the demand go away by then, which is the moment that used to + /// abort the group. + #[tokio::test(start_paused = true)] + async fn ended_demand_finishes_the_open_live_group() { + let source = source_catalog(320, 240); + let source_demand = source._track.demand(); + // Parks the rung's own fetch of the group below, so that fetch is demand + // for as long as the caller holds it and nothing else serves it. + let _source_fetches = source._track.dynamic(); + + // Two frames, so the output's largest moves once both are encoded. + let mut group = source._track.create_group(0u64.into()).unwrap(); + let mut encoder = moq_video::encode::Encoder::new(&{ + let mut config = moq_video::encode::Config::new(320, 240, moq_video::Rate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + config + }) + .unwrap(); + encoder.cut().unwrap(); + let gray = vec![0x80u8; 320 * 240 * 4]; + for timestamp in [0, 33_333] { + for encoded in encoder.encode(&gray_frame(&gray, timestamp)).unwrap() { + hang::container::Frame { + timestamp: encoded.timestamp, + payload: encoded.payload, + } + .write_to(&mut group) + .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 only demand on the rung. Sequence 7 is one the live path never + // reaches, so it parks in the source handler while the live path claims + // output group 0 and encodes both frames of the open source group. + let rung = consumer.track("video/120p").unwrap(); + let fetching = rung.fetch_group(7, None); + let mut edge = rung.clone(); + while edge.largest().await.map(|position| (position.group, position.frame)) != Some((0, 1)) {} + + // Read straight from the cache while the group is still being written. + let mut fetched = rung.fetch_group(0, None).await.unwrap(); + + // The only demand ends with output group 0 still open. + drop(fetching); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + assert!(source_demand.is_used(), "the live path left its source mid-group"); + + // Once the source group ends, the live path finishes its copy and leaves. + group.finish().unwrap(); + source_demand.unused().await.unwrap(); + + let mut frames = 0; + while fetched.read_frame().await.expect("group 0 was aborted").is_some() { + frames += 1; + } + assert_eq!(frames, 2, "group 0 lost frames"); + + transcoder.abort(); + } + /// Two transcoders fed the same source publish groups that mirror the /// source's sequences and timestamps rather than anything numbered per /// instance. @@ -1578,9 +1671,7 @@ mod tests { }) .await; - // A held query keeps the live path encoding across both fetches. let rung = consumer.track("video/120p").unwrap(); - let _query = rung.query(); let mut groups = Vec::new(); for sequence in 0..2 { let mut fetched = rung.fetch_group(sequence, None).await.unwrap(); diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 513d34094ed..7a451b7780d 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -3,9 +3,11 @@ //! Nothing is encoded until someone asks, via the two demand paths moq-net //! exposes on the output track: //! -//! - A live subscription (`used`) starts a live loop that subscribes to the -//! source track (mirroring the aggregate subscription) and transcodes group -//! for group until the track goes `unused` again. +//! - Any demand (`used`), including a fetch, starts a live loop that subscribes +//! to the source track (mirroring the aggregate subscription) and transcodes +//! group for group until the track goes `unused` again. The group open at +//! that point is finished first, so a later reader of that sequence gets the +//! whole group rather than an aborted one. //! - A fetch of a specific group (`requested_group`) fetches that same group //! 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 @@ -208,20 +210,40 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result = None; + // Set while demand is gone but a group is still open. A group name always + // means the same content, so the group is finished for whoever reads that + // sequence later instead of aborted, and the session ends at its boundary. + let mut leaving = false; 'session: loop { let item = tokio::select! { item = listener.recv() => item, - _ = demand.unused() => { + unused = demand.unused(), if !leaving => { + if unused.is_ok() && current.is_some() { + leaving = true; + continue; + } + // Between groups, or the track closed: nothing to finish. + if let Some(output) = current.take() { + output.abort(moq_net::Error::Cancel)?; + } + break 'session; + } + used = demand.used(), if leaving => { + if used.is_ok() { + // Back before the group ended: carry on as if never gone. + leaving = false; + continue; + } + // The track closed under the group. if let Some(output) = current.take() { - // Signal downstream that the group is incomplete. output.abort(moq_net::Error::Cancel)?; } break 'session; @@ -239,7 +261,7 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { - if retiring { + if retiring || leaving { // The group we were riding out never ended, so it is // incomplete however long we wait. if let Some(output) = current.take() { @@ -302,7 +324,7 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result Result Date: Mon, 19 Oct 2026 02:11:31 -0700 Subject: [PATCH 3/5] fix(transcode): recheck returning demand at the live group boundary A select that picked a feed item over the `used` branch, or demand that returned during the encoder flush at `Item::End`, left `leaving` stale and ended the session anyway, so a returning viewer could miss the next group while a new session attached. Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/rung.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 7a451b7780d..3ebe29b51e1 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -259,6 +259,11 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { if retiring || leaving { @@ -324,6 +329,8 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result Date: Mon, 19 Oct 2026 11:14:22 -0700 Subject: [PATCH 4/5] fix(transcode): open no live group once demand is gone The post-item demand recheck only cleared `leaving`, so demand that ended during the encoder flush or lost the select to a queued Group let the live path open that group and ride it out with nobody asking. Snapshot demand both ways, and leave on a stray frame with no group open. Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/lib.rs | 88 ++++++++++++++++++++++++++++++++++++ rs/moq-transcode/src/rung.rs | 21 ++++++--- 2 files changed, 102 insertions(+), 7 deletions(-) diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index 6b71197d2f4..4d3c992112f 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -1629,6 +1629,94 @@ mod tests { transcoder.abort(); } + /// Demand that ends between groups never opens another one. Here the next + /// source group is already queued for the live path when the last fetch + /// ends, so the live path sees both at once. It leaves its source rather + /// than opening that group and riding it out with nobody asking. + /// + /// The live path picks between the two at random, so without the fix this + /// fails about half the rounds, so a run of several fails almost every time. + #[tokio::test(start_paused = true)] + async fn ended_demand_opens_no_new_live_group() { + for _ in 0..8 { + let source = source_catalog(320, 240); + let source_demand = source._track.demand(); + // Parks the rung's own fetch of the group below, so that fetch is + // demand for as long as the caller holds it and nothing else serves it. + let _source_fetches = source._track.dynamic(); + + let mut encoder = moq_video::encode::Encoder::new(&{ + let mut config = moq_video::encode::Config::new(320, 240, moq_video::Rate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + config + }) + .unwrap(); + let gray = vec![0x80u8; 320 * 240 * 4]; + let mut write = |group: &mut moq_net::group::Producer, timestamps: &[u64]| { + encoder.cut().unwrap(); + for ×tamp in timestamps { + for encoded in encoder.encode(&gray_frame(&gray, timestamp)).unwrap() { + hang::container::Frame { + timestamp: encoded.timestamp, + payload: encoded.payload, + } + .write_to(group) + .unwrap(); + } + } + }; + let mut group = source._track.create_group(0u64.into()).unwrap(); + write(&mut group, &[0, 33_333]); + + 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 only demand on the rung, parked on a sequence the live path never + // reaches. Held until the live path has finished group 0. + let rung = consumer.track("video/120p").unwrap(); + let fetching = rung.fetch_group(7, None); + let mut edge = rung.clone(); + while edge.largest().await.map(|position| (position.group, position.frame)) != Some((0, 1)) {} + let mut fetched = rung.fetch_group(0, None).await.unwrap(); + group.finish().unwrap(); + while fetched.read_frame().await.expect("group 0 was aborted").is_some() {} + + // Queue the next source group first, so it is waiting for the live + // path by the time it sees the demand go. + let mut next = source._track.create_group(1u64.into()).unwrap(); + write(&mut next, &[66_666]); + drop(fetching); + + tokio::time::timeout(std::time::Duration::from_secs(10), source_demand.unused()) + .await + .expect("the live path opened a group nobody asked for") + .unwrap(); + + transcoder.abort(); + } + } + /// Two transcoders fed the same source publish groups that mirror the /// source's sequences and timestamps rather than anything numbered per /// instance. diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 3ebe29b51e1..1571b06e575 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -259,10 +259,11 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { @@ -309,8 +310,14 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result { // No open group: attached mid-group, skipped a duplicate, or - // recovering from a lag. Wait for the next boundary. - let Some(output) = &mut current else { continue }; + // recovering from a lag. Wait for the next boundary, unless + // nobody is left to serve it. + let Some(output) = &mut current else { + if leaving { + break 'session; + } + continue; + }; // The feed decodes at the source's native size; size this rung's copy // here. Supported GPU paths feed the encoder without touching the @@ -330,7 +337,7 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result Date: Mon, 19 Oct 2026 11:32:06 -0700 Subject: [PATCH 5/5] fix(transcode): recheck demand before opening a live group Demand that left while the encoder flushed or cut at a group boundary still opened the next output group, which the live path then rode out with nobody asking. Recheck right before create_group. Co-Authored-By: Claude Opus 5.5 --- rs/moq-transcode/src/rung.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index 1571b06e575..3235f2e3af8 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -290,6 +290,11 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result