diff --git a/quest/m1/README.md b/quest/m1/README.md index ed21ac9bf1c..f6fc2880d10 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 - [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 - [Remove what dominates build time](/quest/m1/prune-build-time.md) - the crates, features, and code that cost the most build time are removed or made cheaper diff --git a/quest/m1/fmp4-refusal-framing.md b/quest/m1/fmp4-refusal-framing.md deleted file mode 100644 index ca06ec8e596..00000000000 --- a/quest/m1/fmp4-refusal-framing.md +++ /dev/null @@ -1,43 +0,0 @@ -# [S] fMP4 refusals keep framing - -## 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. - -## 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. - -## 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 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 diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 5f8c1723e52..608b1afd7e1 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,6 +2104,45 @@ mod tests { assert!(import.video[0].importer.is_none(), "the dead rendition is retired"); } + /// 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] { + 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(); + 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); + 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..e03735655c2 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, 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 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)) `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..93cd3c8ec67 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, @@ -302,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(()) } @@ -349,10 +356,28 @@ 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 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<()> { + 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(moq_net::Error::MalformedTrack); + } + 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 +418,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 +445,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 +463,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"); - } return Err(err); } let published = mdat.chunks.is_empty(); @@ -1698,7 +1698,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(); @@ -1765,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(()); @@ -1797,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(()); @@ -1973,12 +1980,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 +1997,33 @@ 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}" + ); + 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!( + 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 +2039,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 +2105,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 +2249,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 +2263,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); } 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) }