diff --git a/quest/m1/README.md b/quest/m1/README.md index 6d724504d5a..492020cee7c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -187,7 +187,6 @@ per-frame CPU growth since #5926. - [Binding announcements always carry captures](/quest/m1/ffi-required-captures.md) - moq-ffi and its generated bindings make `captures` a required list, as #7266 always sets it - [HLS export follows a resolution change](/quest/m1/hls-reinit-same-codec.md) - a decoder-identity change under the same codec string starts a new init and segment tag - [fMP4 decode times keep increasing](/quest/m1/fmp4-bframe-decode-time.md) - fragments shorter than a GOP of B-frame video never step `tfdt` backwards -- [A truncated MKV ends its tracks at finish](/quest/m1/mkv-truncated-finish.md) - `finish()` refusing truncated input still ends the tracks, so subscribers stop waiting - [moq-quic deletes its remaining 0-RTT code](/quest/m1/quic-zero-rtt-leftovers.md) - only the code that drops a 0-RTT packet remains, with a test - [Gate the no-em-dash rule](/quest/m1/em-dash-gate.md) - `just check` fails on an em dash in any tracked text file - [Every package shares one minor version](/quest/m1/shared-version.md) - a recipe sets every published package to one minor at the cut (next: 0.22.0), and `just check` enforces it and the `0.0.x` rule for experiments diff --git a/quest/m1/mkv-truncated-finish.md b/quest/m1/mkv-truncated-finish.md deleted file mode 100644 index a11ea1e5c73..00000000000 --- a/quest/m1/mkv-truncated-finish.md +++ /dev/null @@ -1,19 +0,0 @@ -# [XS] A truncated MKV ends its tracks at finish - -## Goal - -When the MKV importer's `finish()` refuses truncated input (`Mkv(Truncated)`), it still ends its tracks with what was accepted, as it does after a refused `decode`, so a subscriber's `read()` returns instead of waiting until the importer is dropped. `finish()` still returns the error, and the frame cut short still fails its group with the reason. - -## Plan - -Found by the verification in [moq-dev/moq#7515](https://github.com/moq-dev/moq/pull/7515), which fed the live `opus.webm` truncated at every length and with every byte flipped: 1462 truncation lengths and 29 byte flips ended with tracks neither finished nor aborted after `finish()` returned `Truncated`, at every chunk size. Callers that drop the importer after an error were unaffected. - -In `Import::finish` (`rs/moq-mux/src/container/mkv/import.rs`, about lines 838-852), a failed `drain_eof` aborts only the streaming block's group and returns early, skipping the per-track finish. A `decode` that already failed (`self.failed`) skips `drain_eof` and still reaches the track loop, so it ends its tracks. - -Decided 2026-11-01 (maintainer, follow-up planning after the 2026-11-01 /quest-afk run): m1, a small fix in an importer the cut ships. End the tracks on the truncation path the way a refused `decode` already does (`decode`'s docs: "`finish` ends the tracks with what was accepted"), so the two refusals behave alike. #7515 also floated aborting every track with the truncation error instead; matching the refused `decode` keeps one rule and keeps the groups that did arrive. Test with #7515's harness shape: a truncated `opus.webm` whose subscriber's read ends after `finish()` while the importer is still held. - -Public API: none. Wire: none. - -## Related - -- [Untrusted media parsers have fuzz targets](/quest/m2/media-fuzz.md) - could assert that every track ends after `finish()`, whatever the input diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 431ec0720cf..4d301b2badd 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -834,22 +834,24 @@ impl Import { /// Finish all tracks, flushing current groups. /// /// Input that ends inside an element other than an unknown-size Segment or Cluster is refused as - /// truncated. + /// truncated, but the tracks still end with what was accepted. pub fn finish(&mut self) -> Result<()> { self.finished = true; - if self.failed.is_none() - && let Err(err) = self.drain_eof() - { + let mut result = match self.failed { + None => self.drain_eof(), + Some(_) => Ok(()), + }; + if let Err(err) = &result { // A frame cut short fails its group with the reason, rather than waiting for the rest. self.abort_block(err.reason()); - return Err(err); } // No frame follows to anchor the clock, so publish the declared track set now. self.initial_reservation = None; for track in self.tracks.values_mut() { - track.track.finish()?; + // Finish every track even on refusal, keeping the first error. + result = result.and(track.track.finish()); } - Ok(()) + result } /// Abort all tracks with `err` instead of finishing, so subscribers see the real @@ -1801,17 +1803,87 @@ mod tests { import.finish().unwrap(); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn truncation_fails_the_frame_cut_short() { + use futures::FutureExt; + let (mut import, catalog, consumer) = import(); import.decode(&live(vec![entry(1, 1, "V_VP9")])).unwrap(); import.decode(&simple_block(1, 1024)).unwrap(); import.decode(&[4; 512]).unwrap(); let mut frame = first_frame(&catalog, &consumer).await; - assert!(matches!(import.finish(), Err(crate::Error::Mkv(Error::Truncated)))); + let err = import.finish().unwrap_err(); + assert!(matches!(err, crate::Error::Mkv(Error::Truncated))); // The importer is still owned, and the frame fails rather than waiting on it. - let read = kio::clock::timeout(WAIT, frame.read_all()).await.unwrap(); - assert!(read.is_err(), "a truncated frame must not read as complete: {read:?}"); + let read = frame.read_all().now_or_never().expect("the cut-short frame must fail"); + assert!(matches!( + read, + Err(moq_net::Error::Stream(moq_net::StreamError::Internal)) + )); + drop(import); + } + + #[tokio::test(start_paused = true)] + async fn truncated_finish_ends_tracks_with_accepted_content() { + use futures::FutureExt; + + let data = include_bytes!("test_data/opus.webm"); + // The last packet is followed by 10 bytes of BlockGroup metadata: cut inside its payload, + // or inside the metadata after all six packets were accepted. + for (missing, expected) in [(11, 5), (1, 6)] { + let data = &data[..data.len() - missing]; + for chunk in [1, 7, 64, data.len()] { + let (mut import, catalog, consumer) = import(); + for piece in data.chunks(chunk) { + import.decode(piece).unwrap(); + } + assert!(matches!(import.finish(), Err(crate::Error::Mkv(Error::Truncated)))); + let snapshot = catalog.snapshot(); + let (name, config) = snapshot.audio.renditions.iter().next().unwrap(); + let subscription = + moq_net::track::Subscription::default().with_max_delay(std::time::Duration::from_secs(1)); + let track = consumer.track(name).unwrap().subscribe(subscription).await.unwrap(); + let mut reader = + crate::container::Consumer::new(track, crate::catalog::hang::Container::try_from(config).unwrap()); + let mut frames = 0; + while reader + .read_frame() + .now_or_never() + .expect("finish must end the track while the importer is held") + .unwrap() + .is_some() + { + frames += 1; + } + assert_eq!(frames, expected, "chunk {chunk}"); + drop(import); + } + } + } + + #[tokio::test(start_paused = true)] + async fn truncated_finish_publishes_and_ends_empty_tracks() { + let (mut import, catalog, consumer) = import(); + let mut clocks = crate::container::test_util::Clocks::subscribe(&consumer).await; + import + .decode(&live(vec![entry(1, 1, "V_VP9"), entry(2, 1, "V_VP8")])) + .unwrap(); + import.decode(&[0xec, 0x81]).unwrap(); + assert!(clocks.drain().is_empty(), "no block has anchored the catalog yet"); + assert!(matches!(import.finish(), Err(crate::Error::Mkv(Error::Truncated)))); + assert!( + !clocks.drain().is_empty(), + "finish must release the initial catalog reservation" + ); + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.video.renditions.len(), 2); + for name in snapshot.video.renditions.keys() { + let mut track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + assert!(matches!( + track.poll_recv_group(&kio::Waiter::noop()), + std::task::Poll::Ready(Ok(None)) + )); + } drop(import); }