diff --git a/quest/m1/README.md b/quest/m1/README.md index 18f4bd1d617..c74d1432f66 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -103,7 +103,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 - [Random-loss audit](/quest/m1/random-loss-audit.md) - every decision random-loss gates drove is kept, reworked, or reverted on shaped paths 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..4d3c992112f 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,186 @@ 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(); + } + + /// 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. @@ -1578,9 +1759,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..3235f2e3af8 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; @@ -237,9 +259,15 @@ 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() { @@ -262,6 +290,11 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result 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 @@ -302,7 +341,9 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result Result