From d4579e126c4e94fcb45f92c6eb233f7acca89b4a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 03:51:05 -0800 Subject: [PATCH 1/5] chore(quest): claim quest/m1/fmp4-refusal-framing.md Quest-Claim: 1184034e567f919e From 46b7c94b4ca7ef928d3217f386b997739fdb027a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 04:03:41 -0800 Subject: [PATCH 2/5] fix(mux)!: make fMP4 decode refusals terminal Choose terminal refusal over input draining so arbitrary chunking cannot silently resume after dropped bytes. Verify HLS generation recovery and checkpoint the outstanding quiet-host benchmark and full-gate limitation. Co-Authored-By: GPT-6.1 Sol --- quest/m1/fmp4-refusal-framing.md | 79 +++++++----- rs/moq-hls/src/import.rs | 40 ++++++ rs/moq-mux/MIGRATE.md | 6 + rs/moq-mux/src/container/fmp4/import.rs | 129 ++++++++++--------- rs/moq-mux/src/container/fmp4/import_test.rs | 81 ++++++------ 5 files changed, 200 insertions(+), 135 deletions(-) diff --git a/quest/m1/fmp4-refusal-framing.md b/quest/m1/fmp4-refusal-framing.md index ca06ec8e596..73bc2fecf73 100644 --- a/quest/m1/fmp4-refusal-framing.md +++ b/quest/m1/fmp4-refusal-framing.md @@ -1,43 +1,54 @@ -# [S] fMP4 refusals keep framing +# [S] Re-measure interleaved fMP4 import ## Goal -A box the fMP4 importer refuses at its header no longer leaves a caller -feeding arbitrary chunks out of frame. Either the importer reads the rest of -that decode call's input past the refused box, so later calls stay -correctly framed whatever chunk boundaries the caller picks, or the refusal -is terminal and every later call fails, so no input is silently dropped -mid-stream. The 4-track -interleaved import benchmark is no slower than before the streaming change, -or the cost is measured and accepted. +The 4-track, 16-run, whole-input `cmaf-interleaved-import` benchmark is no +slower than before the streaming change, or the cost is measured and accepted. +The quiet-host comparison remains outstanding. ## Plan -Decided 2026-10-31 (maintainer, the /quest-complete requeue session), from -the follow-ups of [moq-dev/moq#7232](https://github.com/moq-dev/moq/pull/7232) -(fMP4 streams mdat) and [moq-dev/moq#7234](https://github.com/moq-dev/moq/pull/7234) -(MKV streams blocks). Both have landed. #7234 landed first, so #7232 already -streams the importer's samples through `Producer::open`/`commit`; the -original open/commit bullet is done, and this quest keeps the other two -([noted on moq-dev/moq#7468](https://github.com/moq-dev/moq/pull/7468#issuecomment-6554670247)). - -- **A refused box.** `Import::decode` (`rs/moq-mux/src/container/fmp4/import.rs`) - documents it: a refused box or fragment discards the rest of its body, and - the rest of the refusing call's input is not read. That suits box-aligned - callers (HLS ingest feeds one segment per call) but loses framing for - arbitrary chunking. Either continue past the refused box, or make a - refusal terminal for the importer as the MKV importer's is. Choose during - implementation the option that keeps the importer's state simpler while - meeting HLS ingest's needs, and record the choice in the PR. -- **Benchmark.** #7232 measured a possible ~25% slowdown in the 4-track - 16-run whole-input `cmaf-interleaved-import` case, in a noisy run where the - 4-track cases swung ±40% between runs of one source. Re-measure it with the - paired comparison on a quiet host, against the commit before #7232, and - fix it or record why it is accepted. - -Public API: none expected. Wire: none. +The framing part is implemented in this checkpoint: a decode error is terminal, +and every later decode returns the first error without reading input. This +removes recovery/skip state instead of draining arbitrary input after an error. +`finish()` still ends accepted tracks without resetting the error. HLS replaces +importers on init-map changes or rendition eviction, while CLI chunked input +propagates a decode error and aborts. MKV leaf limits remain independent. + +The original decision (maintainer, 2026-10-31, /quest-complete requeue) allowed +either draining the refusing call or terminal refusal, whichever kept the state +simpler while meeting HLS needs. It came from +[moq-dev/moq#7232](https://github.com/moq-dev/moq/pull/7232) (fMP4 streams mdat), +[moq-dev/moq#7234](https://github.com/moq-dev/moq/pull/7234) (MKV streams blocks), +and [moq-dev/moq#7468](https://github.com/moq-dev/moq/pull/7468#issuecomment-6554670247). + +- Compare with `038ecfa6b3c69cabf55a7b2bb975c566381ccb3b`, the parent of the + streaming change. The original possible ~25% slowdown came from a noisy run + whose 4-track cases swung ±40% between repeats of the same source. +- This attempt's host is `desktop`, Linux x86_64, 24 allowed logical CPUs. + On 2026-11-02 at 11:53 UTC its load averages were 7.28 / 3.75 / 3.18, + with other agents active. This does not satisfy the quiet-host prerequisite. + Do not treat shared-host timing as acceptance or wait indefinitely for quiet. +- A diagnostic comparison on that shared host used three alternating rounds, + 1 s warmup, 2 s measurement and 30 samples, pinned to performance CPUs 0-7. + The historical whole-input ID lacks today's `-4232444chunk` suffix, so the + matching cases were paired explicitly rather than trusting the harness's + initial "new case" result. Base/current means in milliseconds were + 7.992/1.037, 7.539/1.066 and 6.373/1.034. One-minute load ranged from + 2.42 to 5.44 with other agents and local checks active. These are full-revision + timings, not an isolated attribution to streaming or acceptance of the cost. +- Re-run the paired comparison on a quiet host, then fix a confirmed regression + or ask the maintainer to accept the measured cost. +- The full `just ci` attempt also hit mbx `Argument list too long (os error 7)` + in Rustdoc and the Clippy-gate test. The existing fix belongs to + [moq-dev/moq#7594](https://github.com/moq-dev/moq/pull/7594); no launcher + bypass or upstream mbx change is part of this checkpoint. Re-run that gate + once its owner restores the environment. + +Public API: decode errors are now terminal, including through `import::Container`; +replace the importer to resume. Signatures and wire formats are unchanged. ## Related -- [MKV leaf element limits](/quest/m1/mkv-leaf-limits.md) - the other leftover of the importer streaming work -- [Media fuzz](/quest/m2/media-fuzz.md) - would catch a dropped-input bug like the refused-box one +- [MKV leaf element limits](/quest/m1/mkv-leaf-limits.md) - independent importer work +- [Media fuzz](/quest/m2/media-fuzz.md) - coverage for dropped-input bugs diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 5f8c1723e52..2ae50edb6c4 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -2099,6 +2099,46 @@ mod tests { assert!(import.video[0].importer.is_none(), "the dead rendition is retired"); } + /// A decode refusal poisons that importer even if the source later fixes its segment. + /// HLS recovers by retiring the generation and importing the live window under fresh names. + #[tokio::test] + async fn a_refused_segment_recovers_through_a_new_importer() { + for passthrough in [false, true] { + let dir = temp_dir(); + write_synth(&dir, &synth(true), "video", &[5_000_000, 7_000_000]); + let playlist = + "#EXTM3U\n#EXT-X-TARGETDURATION:2\n#EXT-X-MAP:URI=\"videoinit.mp4\"\n#EXTINF:1,\nvideo0.m4s\n"; + std::fs::write(dir.join("media.m3u8"), playlist).unwrap(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = CatalogProducer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let mut cfg = Config::new(Url::from_file_path(dir.join("media.m3u8")).unwrap()); + cfg.passthrough = passthrough; + let mut import = Import::new(broadcast, catalog.clone(), cfg).unwrap(); + import.init().await.unwrap(); + let previous = catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + + let segment = std::fs::read(dir.join("video1.m4s")).unwrap(); + // An mdat without a moof is refused at its header. + std::fs::write(dir.join("video1.m4s"), b"\0\0\0\x0cmdatdata").unwrap(); + std::fs::write(dir.join("media.m3u8"), format!("{playlist}#EXTINF:1,\nvideo1.m4s\n")).unwrap(); + for failures in 1..=MAX_RENDITION_FAILURES { + let step = import.step(OnError::Warn).await.unwrap(); + assert!(step.failed.is_some()); + assert_eq!(import.video[0].importer.is_none(), failures == MAX_RENDITION_FAILURES); + // Fixing the bytes cannot clear the first decode refusal on the same importer. + std::fs::write(dir.join("video1.m4s"), &segment).unwrap(); + } + let step = import.step(OnError::Warn).await.unwrap(); + assert!(step.failed.is_none()); + assert_eq!(step.wrote_segments, 2); + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.video.renditions.len(), 1); + assert_ne!(*snapshot.video.renditions.keys().next().unwrap(), previous); + import.finish().unwrap(); + std::fs::remove_dir_all(dir).unwrap(); + } + } + /// An ended rendition is never fetched again, so losing its playlist while its siblings /// are still running cannot evict the finished rendition. #[tokio::test] diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index 92d25498a72..914a079a8c1 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### fMP4 decode errors are terminal + +After `container::fmp4::Import::decode` fails, every later decode returns that error without reading input, including through `import::Container`. +Replace the importer to resume instead of feeding it another box or segment. `finish()` still ends the tracks with the accepted media and does not reset the error. +HLS already replaces importers when the init map changes or a failing rendition is evicted. + ### A `Delay` sets where a consumer starts ([#7311](https://github.com/moq-dev/moq/pull/7311)) `container::Consumer::set_max_delay(Duration)` is `set_delay(Delay)`, and `with_delay(Delay)` is its builder. diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index ee3cf7603b9..5668f15d83a 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -78,6 +78,9 @@ pub struct Import { buffer: BytesMut, body: Option, + // A refused chunk may contain unread bytes, so no later chunk can resume its framing. + failed: Option, + // A segment boundary (a styp or an explicit `cut()`) waiting to land on the next fragment. pending_cut: bool, @@ -276,6 +279,7 @@ impl Import { broadcast, buffer: BytesMut::new(), body: None, + failed: None, pending_cut: false, segment: 0, declared: false, @@ -349,10 +353,27 @@ impl Import { /// [`MAX_FRAGMENT_SAMPLES`](super::MAX_FRAGMENT_SAMPLES) samples, is refused before it is /// held. /// - /// A refused box or fragment publishes nothing more, and the rest of its body is discarded. - /// The rest of the refused call's input is not read, so input fed in box-aligned chunks (such - /// as one HLS segment per call) resumes at the next call. - pub fn decode(&mut self, mut data: &[u8]) -> Result<()> { + /// A decode error is terminal: later calls return the same error without reading input. + /// The refusing call may leave bytes unread, so even box-aligned callers must replace the + /// importer to resume. [`finish`](Self::finish) still ends the tracks with what was accepted. + pub fn decode(&mut self, data: &[u8]) -> Result<()> { + if let Some(err) = &self.failed { + return Err(err.clone()); + } + let result = self.drain(data); + if let Err(err) = &result { + if let Some(Body::Media(mut mdat)) = self.body.take() { + mdat.abort(err.reason()); + } + self.buffer = BytesMut::new(); + self.moof = None; + self.deferred = Deferred::default(); + self.failed = Some(err.clone()); + } + result + } + + fn drain(&mut self, mut data: &[u8]) -> Result<()> { loop { let Some(body) = self.body.take() else { let needed = if self.buffer.len() >= 4 && self.buffer[..4] == [0, 0, 0, 1] { @@ -393,22 +414,7 @@ impl Import { Ok(Body::Skip(header.size)) } }; - match body { - Ok(body) => self.body = Some(body), - Err(err) => { - // Nothing of a refused box is held or published. Discard its body, including - // what this call carries, so framing resumes at the next box. - let count = header.size.unwrap_or(data.len()).min(data.len()); - let remaining = header.size.map(|size| size - count); - if self.moof.is_some() { - self.moof_span = self.moof_span.saturating_add(count); - } - if remaining != Some(0) { - self.body = Some(Body::Skip(remaining)); - } - return Err(err); - } - } + self.body = Some(body?); continue; }; match body { @@ -435,8 +441,6 @@ impl Import { data = &data[count..]; remaining -= count; if remaining == 0 { - // Release the held body before parsing, so a refused box leaves no partial - // header behind for the next one. let body = std::mem::take(&mut self.buffer).freeze(); self.end_metadata(kind, body, header_size)?; } else { @@ -455,15 +459,7 @@ impl Import { *remaining -= count; } if let Err(err) = written { - // The fragment is refused: its open frames fail, and the rest of its mdat is - // discarded so framing resumes at the next box. - mdat.abort(moq_net::Error::MalformedTrack); - if mdat.remaining != Some(0) { - self.body = Some(Body::Skip(mdat.remaining)); - } - if let Err(settle) = self.settle() { - tracing::warn!(%settle, "failed to apply a control deferred by a refused fMP4 fragment"); - } + mdat.abort(err.reason()); return Err(err); } let published = mdat.chunks.is_empty(); @@ -1698,7 +1694,8 @@ impl Import { /// /// A fragment whose samples are still arriving was truncated: its open frames abort, failing /// their groups, every track still finishes, and this returns - /// [`UnexpectedEof`](mp4_atom::Error::UnexpectedEof). + /// [`UnexpectedEof`](mp4_atom::Error::UnexpectedEof). After a decode error, this ends the + /// accepted media without reading more input; it does not clear that error for later decodes. pub fn finish(&mut self) -> Result<()> { // No input follows, so the buffer that waited for the rest of an atom is done. self.buffer = BytesMut::new(); @@ -1973,12 +1970,9 @@ mod tests { let err = import.decode(&header).unwrap_err(); assert!(matches!(err, crate::Error::Cmaf(Error::BackwardsDataOffset)), "{err:?}"); assert!(consumer.recv_group().now_or_never().is_none()); - // The refused mdat is discarded rather than parsed as boxes. - import.decode(&[0xA5; 8]).unwrap(); - import.decode(&media_header(id, 1024, &[0], 4, Some(4))).unwrap(); - import.decode(&[0xA5; 4]).unwrap(); + assert_eq!(import.decode(&[0xA5; 8]).unwrap_err().to_string(), err.to_string()); import.finish().unwrap(); - assert!(consumer.recv_group().now_or_never().is_some()); + assert!(consumer.recv_group().now_or_never().unwrap().unwrap().is_none()); } } @@ -1993,6 +1987,30 @@ mod tests { header } + /// The refusing call can also carry a prefix of the next box. None of that input may + /// disappear behind a successful decode, regardless of where the caller splits it. + #[test] + fn a_refusal_is_terminal_across_chunk_boundaries() { + let mut data = box_header(b"mdat", Some(4)); + data.extend_from_slice(&[0xA5; 4]); + data.extend_from_slice(&box_header(b"styp", Some(0))); + for split in 8..=data.len() { + let mut import = importer(); + let first = import.decode(&data[..split]).unwrap_err().to_string(); + assert_eq!( + import.decode(&data[split..]).unwrap_err().to_string(), + first, + "split: {split}" + ); + import.finish().unwrap(); + assert_eq!(import.decode(&[]).unwrap_err().to_string(), first); + assert_eq!( + import.decode(&box_header(b"free", Some(0))).unwrap_err().to_string(), + first + ); + } + } + #[test] fn oversized_metadata_is_refused_at_its_header() { for kind in [b"moov", b"moof"] { @@ -2008,12 +2026,9 @@ mod tests { matches!(err, crate::Error::Cmaf(Error::MetadataTooLarge { size: refused, .. }) if refused == size), "{kind:?}: {err:?}" ); - // Its body is discarded as it arrives, across segments. - for _ in 0..256 { - import.cut().unwrap(); - import.decode(&[0xA5; 1024]).unwrap(); - assert!(import.buffer.is_empty(), "{kind:?}: {} held bytes", import.buffer.len()); - } + assert_eq!(import.decode(&[0xA5; 1024]).unwrap_err().to_string(), err.to_string()); + assert!(import.buffer.is_empty()); + assert!(import.body.is_none()); } } } @@ -2077,12 +2092,10 @@ mod tests { ); assert!(import.moof.is_none()); assert!(import.buffer.is_empty()); - // With no moof, the mdat after it is refused and discarded. - let err = import.decode(&box_header(b"mdat", Some(4))).unwrap_err(); - assert!(matches!(err, crate::Error::Cmaf(Error::NoMoof)), "{err:?}"); - import.decode(&[0xA5; 4]).unwrap(); - import.decode(&media_header(id, 0, &[0], 4, Some(4))).unwrap(); - import.decode(&[0xA5; 4]).unwrap(); + assert_eq!( + import.decode(&box_header(b"mdat", Some(4))).unwrap_err().to_string(), + err.to_string() + ); import.finish().unwrap(); } @@ -2223,7 +2236,7 @@ mod tests { } #[test] - fn refused_fragment_discards_its_mdat() { + fn refused_fragment_ends_its_import() { const SIZE: usize = 64; for passthrough in [false, true] { let (mut import, id, _) = initialized(passthrough); @@ -2237,19 +2250,13 @@ mod tests { import .decode(&media_header(id, 0, &[0], SIZE as u32, Some(SIZE))) .unwrap(); - import.decode(&[0xA5; SIZE / 2]).unwrap_err(); - // The rest of the refused mdat is discarded rather than parsed as boxes. - import.decode(&[0xA5; SIZE / 2]).unwrap(); - import - .decode(&media_header(id, 96_000, &[0], SIZE as u32, Some(SIZE))) - .unwrap(); - import.decode(&[0xA5; SIZE]).unwrap(); - import.finish().unwrap(); + let err = import.decode(&[0xA5; SIZE / 2]).unwrap_err(); assert_eq!( - groups(&mut consumer), - [(0, Some(1)), (1, Some(1))], - "passthrough: {passthrough}" + import.decode(&[0xA5; SIZE / 2]).unwrap_err().to_string(), + err.to_string() ); + import.finish().unwrap(); + assert_eq!(groups(&mut consumer), [(0, Some(1))], "passthrough: {passthrough}"); } } } diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 8feb3363323..9e22188f67c 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -2407,7 +2407,7 @@ fn non_advancing_fragment_decode_time_is_rejected() { } #[test] -fn rejected_fragment_preserves_decode_time() { +fn rejected_fragment_keeps_the_first_error() { let (ftyp, moov) = decode_init(include_bytes!("test_data/bbb.mp4")); let mut init = Vec::new(); ftyp.encode(&mut init).unwrap(); @@ -2417,14 +2417,13 @@ fn rejected_fragment_preserves_decode_time() { let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); fmp4.decode(&init).unwrap(); fmp4.decode(&audio_fragment(4096, 1024, 327)).unwrap(); - assert!(fmp4.decode(&audio_fragment(3000, 1024, 327)).is_err()); - assert!(matches!( - fmp4.decode(&audio_fragment(3500, 1024, 327)), - Err(crate::Error::Cmaf( - crate::container::fmp4::Error::NonMonotonicDecodeTime { decode_time, previous, .. } - )) if previous == moq_net::Timestamp::from_scale(4096, 44100).unwrap() - && decode_time == moq_net::Timestamp::from_scale(3500, 44100).unwrap() - )); + let err = fmp4.decode(&audio_fragment(3000, 1024, 327)).unwrap_err(); + for time in [3500, 5120] { + assert_eq!( + fmp4.decode(&audio_fragment(time, 1024, 327)).unwrap_err().to_string(), + err.to_string() + ); + } } /// Like [`audio_fragment`] but carrying `samples` samples, the shape a real segmenter emits. @@ -2829,28 +2828,30 @@ async fn seek_keeps_the_decode_time_until_a_discontinuity() { async fn seek_keeps_the_decode_time_until_a_discontinuity_in(passthrough: bool) { let (init, _, _) = bbb_init(); - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); - let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()).with_passthrough(passthrough); - fmp4.decode(&init).unwrap(); - fmp4.decode(&live_session(5_000_000, 5)).unwrap(); - - fmp4.seek(100).unwrap(); - let err = fmp4.decode(&live_session(0, 5)).unwrap_err(); - assert!( - matches!( - err, - crate::Error::Cmaf(crate::container::fmp4::Error::NonMonotonicDecodeTime { .. }) - ), - "{err:?}" - ); - - fmp4.discontinuity().unwrap(); - fmp4.seek(200).unwrap(); - let err = fmp4.decode(&live_session(0, 5)).unwrap_err(); - // Passthrough writes its groups by hand, so it refuses the rewound group itself, the way the - // media producer refuses an unwrapped track's rewound keyframe. - assert!(matches!(err, crate::Error::TimestampRewind(_)), "{err:?}"); + for discontinuity in [false, true] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()).with_passthrough(passthrough); + fmp4.decode(&init).unwrap(); + fmp4.decode(&live_session(5_000_000, 5)).unwrap(); + if discontinuity { + fmp4.discontinuity().unwrap(); + } + fmp4.seek(100).unwrap(); + let err = fmp4.decode(&live_session(0, 5)).unwrap_err(); + if discontinuity { + // Passthrough and unwrapped tracks both refuse a rewound group themselves. + assert!(matches!(err, crate::Error::TimestampRewind(_)), "{err:?}"); + } else { + assert!( + matches!( + err, + crate::Error::Cmaf(crate::container::fmp4::Error::NonMonotonicDecodeTime { .. }) + ), + "{err:?}" + ); + } + } } /// The catalog is first published at the first fragment, carrying the clock that fragment @@ -3221,7 +3222,7 @@ fn empty_and_multiple_webvtt_cues_are_supported() { } /// A text sample is staged whole before it is rewritten, so one past the bound is refused before -/// any of its bytes arrive, and the rest of its mdat is discarded. +/// any of its bytes arrive, and later input stays refused. #[test] fn oversized_text_sample_is_refused_before_staging() { let size = super::MAX_HELD_BYTES + 1; @@ -3265,15 +3266,15 @@ fn oversized_text_sample_is_refused_before_staging() { matches!(err, crate::Error::Cmaf(super::Error::TextSampleTooLarge(refused)) if refused == size), "{err:?}" ); - let chunk = vec![0xA5; 1 << 20]; - let mut remaining = size; - while remaining > 0 { - let count = remaining.min(chunk.len()); - importer.decode(&chunk[..count]).unwrap(); - remaining -= count; - } + assert_eq!(importer.decode(&[0xA5; 1024]).unwrap_err().to_string(), err.to_string()); let cue = subtitle_box(b"vttc", &subtitle_box(b"payl", b"hello")); - importer.decode(&webvtt_fragment(0, &[cue], 1000)).unwrap(); + assert_eq!( + importer + .decode(&webvtt_fragment(0, &[cue], 1000)) + .unwrap_err() + .to_string(), + err.to_string() + ); importer.finish().unwrap(); assert_eq!(catalog.snapshot().text.renditions.len(), 1); } From 3ed95cc36a88419124b510fddfdaad192a27c3c1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 04:11:36 -0800 Subject: [PATCH 3/5] fix(hls): replace refused fMP4 importers immediately Keep failed importers inert, preserve malformed-track abort codes, and document terminal decoding on the shared container surface. Record the passing clean-shell gate separately from the inherited-environment failure. Co-Authored-By: GPT-6.1 Sol --- quest/m1/fmp4-refusal-framing.md | 23 +++++++++++++---------- rs/moq-hls/src/import.rs | 22 ++++++++++++---------- rs/moq-mux/MIGRATE.md | 4 ++-- rs/moq-mux/src/container/fmp4/import.rs | 19 ++++++++++++++++--- rs/moq-mux/src/import/container.rs | 2 ++ 5 files changed, 45 insertions(+), 25 deletions(-) diff --git a/quest/m1/fmp4-refusal-framing.md b/quest/m1/fmp4-refusal-framing.md index 73bc2fecf73..4ddcacb5b80 100644 --- a/quest/m1/fmp4-refusal-framing.md +++ b/quest/m1/fmp4-refusal-framing.md @@ -9,10 +9,12 @@ The quiet-host comparison remains outstanding. ## Plan The framing part is implemented in this checkpoint: a decode error is terminal, -and every later decode returns the first error without reading input. This +and later decodes and controls return the first error without reading input or +changing tracks. This removes recovery/skip state instead of draining arbitrary input after an error. -`finish()` still ends accepted tracks without resetting the error. HLS replaces -importers on init-map changes or rendition eviction, while CLI chunked input +`finish()` still ends accepted tracks without resetting the error. HLS retires +a refused importer immediately and replaces it on the next pass, while transient +fetch failures keep their retry budget. CLI chunked input propagates a decode error and aborts. MKV leaf limits remain independent. The original decision (maintainer, 2026-10-31, /quest-complete requeue) allowed @@ -39,13 +41,14 @@ and [moq-dev/moq#7468](https://github.com/moq-dev/moq/pull/7468#issuecomment-655 timings, not an isolated attribution to streaming or acceptance of the cost. - Re-run the paired comparison on a quiet host, then fix a confirmed regression or ask the maintainer to accept the measured cost. -- The full `just ci` attempt also hit mbx `Argument list too long (os error 7)` - in Rustdoc and the Clippy-gate test. The existing fix belongs to - [moq-dev/moq#7594](https://github.com/moq-dev/moq/pull/7594); no launcher - bypass or upstream mbx change is part of this checkpoint. Re-run that gate - once its owner restores the environment. - -Public API: decode errors are now terminal, including through `import::Container`; +- Inherited-environment `just ci` hit mbx `Argument list too long (os error 7)` + in Rustdoc and the Clippy-gate test. The authorized clean-shell command, + `quest gate nix develop --ignore-environment --keep HOME --keep USER --keep PATH --command just ci origin/main`, + passed all selected checks while retaining mbx. No launcher bypass or upstream + mbx change is part of this checkpoint. + +Public API: decode errors are now terminal for decoding and controls, including +through `import::Container`; replace the importer to resume. Signatures and wire formats are unchanged. ## Related diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 2ae50edb6c4..a15372e7966 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -680,7 +680,12 @@ impl TrackState { rendition: self.label.clone(), source: err, }, - err => err.into(), + err => { + // A decode error is terminal. Retire this generation now instead of fetching + // another segment for an importer that cannot read it. + self.evict(); + err.into() + } })?; self.media_range = range; @@ -2099,8 +2104,8 @@ mod tests { assert!(import.video[0].importer.is_none(), "the dead rendition is retired"); } - /// A decode refusal poisons that importer even if the source later fixes its segment. - /// HLS recovers by retiring the generation and importing the live window under fresh names. + /// A decode refusal retires its importer at once. HLS can import a repaired source on the + /// next pass under fresh track names, without spending the transient fetch retry budget. #[tokio::test] async fn a_refused_segment_recovers_through_a_new_importer() { for passthrough in [false, true] { @@ -2121,13 +2126,10 @@ mod tests { // An mdat without a moof is refused at its header. std::fs::write(dir.join("video1.m4s"), b"\0\0\0\x0cmdatdata").unwrap(); std::fs::write(dir.join("media.m3u8"), format!("{playlist}#EXTINF:1,\nvideo1.m4s\n")).unwrap(); - for failures in 1..=MAX_RENDITION_FAILURES { - let step = import.step(OnError::Warn).await.unwrap(); - assert!(step.failed.is_some()); - assert_eq!(import.video[0].importer.is_none(), failures == MAX_RENDITION_FAILURES); - // Fixing the bytes cannot clear the first decode refusal on the same importer. - std::fs::write(dir.join("video1.m4s"), &segment).unwrap(); - } + let step = import.step(OnError::Warn).await.unwrap(); + assert!(step.failed.is_some()); + assert!(import.video[0].importer.is_none()); + std::fs::write(dir.join("video1.m4s"), &segment).unwrap(); let step = import.step(OnError::Warn).await.unwrap(); assert!(step.failed.is_none()); assert_eq!(step.wrote_segments, 2); diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index 914a079a8c1..e03735655c2 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -6,9 +6,9 @@ How to update your code for each breaking change, newest first. ### fMP4 decode errors are terminal -After `container::fmp4::Import::decode` fails, every later decode returns that error without reading input, including through `import::Container`. +After `container::fmp4::Import::decode` fails, later decodes, cuts, seeks and discontinuities return that error without reading input or changing tracks, including through `import::Container`. Replace the importer to resume instead of feeding it another box or segment. `finish()` still ends the tracks with the accepted media and does not reset the error. -HLS already replaces importers when the init map changes or a failing rendition is evicted. +HLS replaces a refused importer immediately; transient fetch failures still keep their retry budget. ### A `Delay` sets where a consumer starts ([#7311](https://github.com/moq-dev/moq/pull/7311)) diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 5668f15d83a..93cd3c8ec67 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -306,6 +306,9 @@ impl Import { /// it follows the same segmentation as video. A passthrough video track rolls there as well, /// while an unwrapped one opens a group at each sync sample. pub fn cut(&mut self) -> Result<()> { + if let Some(err) = &self.failed { + return Err(err.clone()); + } self.pending_cut = true; Ok(()) } @@ -353,7 +356,8 @@ impl Import { /// [`MAX_FRAGMENT_SAMPLES`](super::MAX_FRAGMENT_SAMPLES) samples, is refused before it is /// held. /// - /// A decode error is terminal: later calls return the same error without reading input. + /// A decode error is terminal: later decodes, cuts, seeks and discontinuities return the same + /// error without reading input or changing the tracks. /// The refusing call may leave bytes unread, so even box-aligned callers must replace the /// importer to resume. [`finish`](Self::finish) still ends the tracks with what was accepted. pub fn decode(&mut self, data: &[u8]) -> Result<()> { @@ -363,7 +367,7 @@ impl Import { let result = self.drain(data); if let Err(err) = &result { if let Some(Body::Media(mut mdat)) = self.body.take() { - mdat.abort(err.reason()); + mdat.abort(moq_net::Error::MalformedTrack); } self.buffer = BytesMut::new(); self.moof = None; @@ -459,7 +463,7 @@ impl Import { *remaining -= count; } if let Err(err) = written { - mdat.abort(err.reason()); + mdat.abort(moq_net::Error::MalformedTrack); return Err(err); } let published = mdat.chunks.is_empty(); @@ -1762,6 +1766,9 @@ impl Import { /// A fragment whose samples are still arriving owns its groups until its last sample, so the /// seek waits for it to finish publishing. pub fn seek(&mut self, sequence: u64) -> Result<()> { + if let Some(err) = &self.failed { + return Err(err.clone()); + } if self.publishing() { self.deferred.seek = Some(sequence); return Ok(()); @@ -1794,6 +1801,9 @@ impl Import { /// rewinds, before publishing any of it. Like [`seek`](Self::seek), it waits for a fragment /// still arriving to finish publishing. pub fn discontinuity(&mut self) -> Result<()> { + if let Some(err) = &self.failed { + return Err(err.clone()); + } if self.publishing() { self.deferred.discontinuity = true; return Ok(()); @@ -2002,6 +2012,9 @@ mod tests { first, "split: {split}" ); + assert_eq!(import.cut().unwrap_err().to_string(), first); + assert_eq!(import.seek(100).unwrap_err().to_string(), first); + assert_eq!(import.discontinuity().unwrap_err().to_string(), first); import.finish().unwrap(); assert_eq!(import.decode(&[]).unwrap_err().to_string(), first); assert_eq!( diff --git a/rs/moq-mux/src/import/container.rs b/rs/moq-mux/src/import/container.rs index e756f3ecfe7..4253f41db03 100644 --- a/rs/moq-mux/src/import/container.rs +++ b/rs/moq-mux/src/import/container.rs @@ -112,6 +112,8 @@ impl Container { } /// Decode a chunk of container bytes. + /// + /// An fMP4 or MKV decode error is terminal: replace the importer before feeding more input. pub fn decode(&mut self, data: &[u8]) -> Result<()> { self.inner.decode(data) } From 7d0717fc4888d72c459d6716a0f1f6b07c3f81a1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 10:27:33 -0800 Subject: [PATCH 4/5] chore(quest): finish quest/m1/fmp4-refusal-framing.md Paired runs on 038ecfa (before #7232), 951cca (#7232), main at 96244dd and this branch: the 4-track 16-run whole-input interleaved import is 79% faster than before the streaming change, and this branch adds no measurable cost there. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/fmp4-refusal-framing.md | 57 -------------------------------- quest/m1/mkv-leaf-limits.md | 4 --- 3 files changed, 62 deletions(-) delete mode 100644 quest/m1/fmp4-refusal-framing.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 63ad9130f00..65083937341 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -135,7 +135,6 @@ per-frame CPU growth since #5926. - [Address groups by timestamp](/quest/m1/time-addressing/README.md) - group start timestamps replace sequence numbers in the moq-net API and lite-07 wire, while old subscribers, old publishers, and published JS clients keep working - [Pre-cut audit findings](/quest/m1/cut-audit/README.md) - fix the crashes, hostile-input costs, and correctness bugs the 2026-10-24 audit of main found - [Vendored code has every upstream fix it needs](/quest/m1/vendor-sync/README.md) - quinn, libv4l-rs, nvidia-video-codec-sdk, rust-media-libs, cros-codecs, and UniFFI generator fixes are ported or planned, once per release cycle -- [fMP4 refusals keep framing](/quest/m1/fmp4-refusal-framing.md) - a box refused at its header stops dropping the rest of its call's input, and the interleaved import benchmark is re-measured - [MKV leaf element limits](/quest/m1/mkv-leaf-limits.md) - `CodecID`, `DocType`, and `CodecPrivate` are refused past per-element size limits instead of read whole - [Delete the tests that don't earn their cost](/quest/m1/prune-tests.md) - an off-cycle test cost sweep deletes tests whose cost outweighs their signal and makes the rest cheaper - [Remove the dependencies that don't earn their weight](/quest/m1/prune-deps.md) - Rust, JS, and vendored dependencies that are trivial, duplicated, or serve features not worth keeping are removed diff --git a/quest/m1/fmp4-refusal-framing.md b/quest/m1/fmp4-refusal-framing.md deleted file mode 100644 index 4ddcacb5b80..00000000000 --- a/quest/m1/fmp4-refusal-framing.md +++ /dev/null @@ -1,57 +0,0 @@ -# [S] Re-measure interleaved fMP4 import - -## Goal - -The 4-track, 16-run, whole-input `cmaf-interleaved-import` benchmark is no -slower than before the streaming change, or the cost is measured and accepted. -The quiet-host comparison remains outstanding. - -## Plan - -The framing part is implemented in this checkpoint: a decode error is terminal, -and later decodes and controls return the first error without reading input or -changing tracks. This -removes recovery/skip state instead of draining arbitrary input after an error. -`finish()` still ends accepted tracks without resetting the error. HLS retires -a refused importer immediately and replaces it on the next pass, while transient -fetch failures keep their retry budget. CLI chunked input -propagates a decode error and aborts. MKV leaf limits remain independent. - -The original decision (maintainer, 2026-10-31, /quest-complete requeue) allowed -either draining the refusing call or terminal refusal, whichever kept the state -simpler while meeting HLS needs. It came from -[moq-dev/moq#7232](https://github.com/moq-dev/moq/pull/7232) (fMP4 streams mdat), -[moq-dev/moq#7234](https://github.com/moq-dev/moq/pull/7234) (MKV streams blocks), -and [moq-dev/moq#7468](https://github.com/moq-dev/moq/pull/7468#issuecomment-6554670247). - -- Compare with `038ecfa6b3c69cabf55a7b2bb975c566381ccb3b`, the parent of the - streaming change. The original possible ~25% slowdown came from a noisy run - whose 4-track cases swung ±40% between repeats of the same source. -- This attempt's host is `desktop`, Linux x86_64, 24 allowed logical CPUs. - On 2026-11-02 at 11:53 UTC its load averages were 7.28 / 3.75 / 3.18, - with other agents active. This does not satisfy the quiet-host prerequisite. - Do not treat shared-host timing as acceptance or wait indefinitely for quiet. -- A diagnostic comparison on that shared host used three alternating rounds, - 1 s warmup, 2 s measurement and 30 samples, pinned to performance CPUs 0-7. - The historical whole-input ID lacks today's `-4232444chunk` suffix, so the - matching cases were paired explicitly rather than trusting the harness's - initial "new case" result. Base/current means in milliseconds were - 7.992/1.037, 7.539/1.066 and 6.373/1.034. One-minute load ranged from - 2.42 to 5.44 with other agents and local checks active. These are full-revision - timings, not an isolated attribution to streaming or acceptance of the cost. -- Re-run the paired comparison on a quiet host, then fix a confirmed regression - or ask the maintainer to accept the measured cost. -- Inherited-environment `just ci` hit mbx `Argument list too long (os error 7)` - in Rustdoc and the Clippy-gate test. The authorized clean-shell command, - `quest gate nix develop --ignore-environment --keep HOME --keep USER --keep PATH --command just ci origin/main`, - passed all selected checks while retaining mbx. No launcher bypass or upstream - mbx change is part of this checkpoint. - -Public API: decode errors are now terminal for decoding and controls, including -through `import::Container`; -replace the importer to resume. Signatures and wire formats are unchanged. - -## Related - -- [MKV leaf element limits](/quest/m1/mkv-leaf-limits.md) - independent importer work -- [Media fuzz](/quest/m2/media-fuzz.md) - coverage for dropped-input bugs diff --git a/quest/m1/mkv-leaf-limits.md b/quest/m1/mkv-leaf-limits.md index 0c9ddfc782b..9a96799e084 100644 --- a/quest/m1/mkv-leaf-limits.md +++ b/quest/m1/mkv-leaf-limits.md @@ -29,7 +29,3 @@ importer's existing error for a refused element. Test each element declared past its limit, fed in small chunks. Public API: none. Wire: none. - -## Related - -- [fMP4 refusals keep framing](/quest/m1/fmp4-refusal-framing.md) - the fMP4 leftover of the same streaming work From da9c13da2a7beec446e28213866a3ec3326ec2ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 11:56:20 -0800 Subject: [PATCH 5/5] docs(hls): name the per-rendition failure count a refusal skips Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/import.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index a15372e7966..608b1afd7e1 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -2104,8 +2104,10 @@ mod tests { assert!(import.video[0].importer.is_none(), "the dead rendition is retired"); } - /// A decode refusal retires its importer at once. HLS can import a repaired source on the - /// next pass under fresh track names, without spending the transient fetch retry budget. + /// A decode refusal retires its importer at once, rather than after + /// [`MAX_RENDITION_FAILURES`] consecutive failures, so the next pass rebuilds it under fresh + /// track names and can import a repaired source. The refusal is not settled, so `run` still + /// retries the step within its backoff budget. #[tokio::test] async fn a_refused_segment_recovers_through_a_new_importer() { for passthrough in [false, true] {