diff --git a/crates/media/README.md b/crates/media/README.md index 8683b124d..398ba2775 100644 --- a/crates/media/README.md +++ b/crates/media/README.md @@ -328,13 +328,20 @@ while let Some(p) = d.next_packet()? { // file order, all tracks } ``` -- **Open** (`open`, `open_with(src, &OpenOptions { index, verify_crc })`): EBML header (DocType +- **Open** (`open`, `open_with(src, &OpenOptions { index, verify_crc, index_limit })`): EBML header (DocType `matroska` or `webm`), the first Segment, and its level-1 elements: SeekHead (followed for elements after the clusters), Info, Tracks, Cues, Chapters, Tags, Attachments (listed with offset and size; payloads are not read). With `index: true` (the default) every Cluster is scanned, element and block headers only, to build per-track `Sample` tables (`offset`, `size`, `pts`, `duration`, `keyframe`, cluster, block offset, lace index, `BlockAdditional` location) and keyframe lists. With `index: false`, `open` stops at the first Cluster. +- **Progressive index** (`index_limit: Some(offset)`): the scan stops at that byte offset, so a + source that has loaded only the front of the file opens at once. The first Cluster is always + scanned; later ones only if they end below the limit. `MkvFile::index_end` is `Some(offset)` + while the index is incomplete (`indexed` is `false`; the tables are exact for everything before + it; Cues after the scanned part are not read yet), and `MkvFile::extend_index(src, &opts)` + continues from there with a larger limit (`None`: to the end, which completes the index) + without rescanning. A read that fails part-way (data not loaded yet) can be retried. - **Timestamps:** Matroska ticks are `TimestampScale` ns (usually 1 ms); `Track::timebase` is that as a reduced fraction of a second. `pts` is in ticks, `pts_ns` in exact nanoseconds. `CodecDelay` is subtracted (RFC 9559), as FFmpeg does, so Opus and Vorbis priming gives negative pts. Laced diff --git a/crates/media/src/matroska/demux.rs b/crates/media/src/matroska/demux.rs index db80f968d..964585449 100644 --- a/crates/media/src/matroska/demux.rs +++ b/crates/media/src/matroska/demux.rs @@ -67,6 +67,18 @@ impl FrameBudget { Self { frames: by_len.clamp(MIN_FRAME_ALLOWANCE, MAX_FRAMES), bytes: file_len } } + /// The budget left for a file whose `tracks` already hold the frames of an earlier scan. + fn resumed(file_len: u64, tracks: &[Track]) -> Self { + let mut b = Self::new(file_len); + let (frames, bytes) = tracks + .iter() + .flat_map(|t| t.samples.iter()) + .fold((0usize, 0u64), |(n, b), s| (n.saturating_add(1), b.saturating_add(u64::from(s.size.max(1))))); + b.frames = b.frames.saturating_sub(frames); + b.bytes = b.bytes.saturating_sub(bytes); + b + } + fn take(&mut self, frames: &[(u64, u32)]) -> Result<()> { self.frames = self.frames.checked_sub(frames.len()).ok_or(Error::TooLarge("frame count"))?; let bytes = frames.iter().map(|&(_, n)| u64::from(n.max(1))).sum::(); @@ -88,11 +100,26 @@ pub struct OpenOptions { /// Verify CRC-32 elements of level-1 elements (and clusters while indexing). Mismatches are /// recorded in [`MkvFile::crc_errors`]; parsing continues. Default `false`. pub verify_crc: bool, + /// Index only the front of the file: stop the cluster scan at this absolute byte offset. + /// Default `None` (scan every cluster). + /// + /// The first Cluster is always scanned (a file needs at least one to be useful). After it, an + /// element of known size (a Cluster, or Cues or Tags between Clusters) is read only if it lies + /// entirely below the limit, so a byte source that has loaded only the front of the file is + /// enough. Two things can still read past the limit: the first Cluster, and a Cluster of + /// unknown size (live streams), whose end is only found by scanning it. Info and Tracks are + /// read as usual; Cues, Tags and the other elements after the scanned part are not read + /// until the index is completed. + /// + /// The sample tables then cover everything before [`MkvFile::index_end`], which is `Some` + /// until the index is complete ([`MkvFile::indexed`]). Continue with [`MkvFile::extend_index`] + /// without rescanning. Ignored when `index` is `false`. + pub index_limit: Option, } impl Default for OpenOptions { fn default() -> Self { - Self { index: true, verify_crc: false } + Self { index: true, verify_crc: false, index_limit: None } } } @@ -123,6 +150,13 @@ pub struct MkvFile { pub first_cluster: Option, /// Sample tables have been built by scanning all clusters. pub indexed: bool, + /// Set while an open limited by [`OpenOptions::index_limit`] left the index incomplete: the + /// absolute offset where the cluster scan stopped and where [`MkvFile::extend_index`] + /// continues. The tables cover every frame before it and none after. `None` once the index + /// is complete, and for files opened without an index. + pub index_end: Option, + /// Walker state to continue the scan from [`MkvFile::index_end`]. + resume: Option, /// Number of CRC-32 elements verified (with `verify_crc`). pub crc_checked: u32, /// Absolute offsets of elements whose CRC-32 did not match. @@ -131,6 +165,14 @@ pub struct MkvFile { pub warnings: Vec, } +/// What a scan stopped at [`MkvFile::index_end`] needs to continue: the level-1 elements already +/// read (so a later pass does not read them twice) and the metadata byte budget left. +#[derive(Clone, Debug, Default)] +struct Resume { + parsed: HashMap, + meta_left: u64, +} + /// One demuxed frame. #[derive(Clone, Debug, PartialEq)] pub struct Packet { @@ -208,6 +250,10 @@ impl Default for Cache { struct Io<'a, S: ByteSource + ?Sized> { src: &'a S, len: u64, + /// Bytes the caller has available (an [`OpenOptions::index_limit`]): the read window does not + /// run past it unless a read needs the bytes, so a source that has loaded only the front of + /// the file is not asked for the rest. `len` without a limit. + avail: u64, cache: &'a mut Cache, } @@ -238,7 +284,7 @@ impl Hdr { impl<'a, S: ByteSource + ?Sized> Io<'a, S> { fn new(src: &'a S, cache: &'a mut Cache) -> Self { let len = src.len(); - Self { src, len, cache } + Self { src, len, avail: len, cache } } /// Up to `n` (≤ WIN) bytes at `pos` (fewer at EOF). @@ -250,7 +296,10 @@ impl<'a, S: ByteSource + ?Sized> Io<'a, S> { let want_end = (pos + n as u64).min(self.len); let c = &*self.cache; if !(pos >= c.start && want_end <= c.start + c.buf.len() as u64) { - let fill = ((self.len - pos) as usize).min(WIN); + let window_end = (pos + WIN as u64).min(self.len); + // stay inside what is available, but never fetch less than this read needs + let end = if self.avail > pos { window_end.min(self.avail).max(want_end) } else { want_end }; + let fill = (end - pos) as usize; let mut buf = std::mem::take(&mut self.cache.buf); buf.resize(fill, 0); self.src.read_at(pos, &mut buf)?; @@ -280,7 +329,8 @@ impl<'a, S: ByteSource + ?Sized> Io<'a, S> { if pos >= limit { return Ok(None); } - let b = self.peek(pos, 12)?; + // never more than the parent holds: its last element may be shorter than a maximal header + let b = self.peek(pos, 12.min(usize::try_from(limit - pos).unwrap_or(12)))?; let Some((id, il)) = read_id(b) else { return Ok(None) }; let Some((size, sl)) = read_size(&b[il..]) else { return Ok(None) }; let hlen = il + sl; @@ -678,12 +728,20 @@ fn open_impl(src: &S, opts: &OpenOptions, cache: &mut Ca // each pass over the file (open, building the index) gets the whole budget cache.recovery_left = RECOVERY_BUDGET; let mut io = Io::new(src, cache); + if let Some(l) = opts.index_limit.filter(|_| opts.index) { + io.avail = l; + } let len = io.len; // EBML header (tolerate leading junk in the first 64 KiB) let start = { - let b = io.peek(0, WIN)?; let magic = [0x1A, 0x45, 0xDF, 0xA3]; - b.windows(4).position(|w| w == magic).ok_or_else(|| Error::NotMatroska("no EBML header".into()))? as u64 + // the usual case needs 4 bytes, not the whole window (a source may hold only the front of the file) + if io.peek(0, 4)? == magic { + 0 + } else { + let b = io.peek(0, WIN)?; + b.windows(4).position(|w| w == magic).ok_or_else(|| Error::NotMatroska("no EBML header".into()))? as u64 + } }; let eh = io.header(start, len)?.ok_or_else(|| Error::NotMatroska("bad EBML header".into()))?; let hdr_end = eh.end(len); @@ -735,6 +793,8 @@ fn open_impl(src: &S, opts: &OpenOptions, cache: &mut Ca attachments: Vec::new(), first_cluster: None, indexed: false, + index_end: None, + resume: None, crc_checked: 0, crc_errors: Vec::new(), warnings: Vec::new(), @@ -742,7 +802,8 @@ fn open_impl(src: &S, opts: &OpenOptions, cache: &mut Ca if seg.overruns(len) { warn(&mut file.warnings, "Segment extends past end of file (truncated)".into()); } - let mut w = Walker { opts, parsed: HashMap::new(), meta_left: len, blocks_seen: false, lookup: None, budget: FrameBudget::new(len) }; + let mut w = + Walker { opts, parsed: HashMap::new(), meta_left: len, blocks_seen: false, resuming: false, lookup: None, budget: FrameBudget::new(len) }; w.walk(&mut io, &mut file)?; finish_tracks(&mut file); Ok(file) @@ -756,6 +817,9 @@ struct Walker<'o> { /// valid file stays within its length; SeekHead entries pointing into one another do not. meta_left: u64, blocks_seen: bool, + /// Continuing a scan (see [`MkvFile::extend_index`]): keep `file.index_end` and the resume + /// state current, as a limited open does, so that a failed read can be retried. + resuming: bool, /// Built at the first Cluster after Tracks is parsed. lookup: Option, budget: FrameBudget, @@ -763,9 +827,42 @@ struct Walker<'o> { impl Walker<'_> { fn walk(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile) -> Result<()> { + let start = file.segment_data_start; + self.walk_from(io, file, start) + } + + /// Walk the Segment's children from `start`. With an [`OpenOptions::index_limit`] the walk may + /// stop early, leaving `file.index_end` and the resume state for [`MkvFile::extend_index`]. + fn walk_from(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, start: u64) -> Result<()> { + let result = self.walk_children(io, file, start); + // Stopped early or failed part-way: hand the elements read so far to the next + // `extend_index`. `parsed` holds only elements read in full (see `level1`), so it is + // moved, not copied per element. + if file.index_end.is_some() { + file.resume = Some(Resume { parsed: std::mem::take(&mut self.parsed), meta_left: self.meta_left }); + } + result + } + + fn walk_children(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, start: u64) -> Result<()> { let seg_end = file.segment_end; - let mut pos = file.segment_data_start; + let limit = self.opts.index_limit.filter(|_| self.opts.index); + if let Some(l) = limit { + io.avail = l; + } + let mut pos = start; + let track = limit.is_some() || self.resuming; + let mut stopped = false; while pos < seg_end { + if track { + self.save(file, pos); + } + // the first Cluster is always scanned; the rest only while they fit below the limit + // (checked before the header is read: that lies past the limit too) + if limit.is_some_and(|l| pos >= l) && !file.clusters.is_empty() { + stopped = true; + break; + } let Some(h) = io.header(pos, seg_end)? else { match io.resync(pos + 1, seg_end)? { Some(p) => { @@ -776,12 +873,20 @@ impl Walker<'_> { None => break, } }; + if let Some(l) = limit + && !file.clusters.is_empty() + && matches!(h.size, Size::Known(_)) + && h.end(seg_end) > l + { + stopped = true; + break; + } match h.id { CLUSTER => { if file.first_cluster.is_none() { file.first_cluster = Some(h.pos); // make sure Info/Tracks are known before interpreting blocks - self.follow_seek_head(io, file, &[INFO, TRACKS])?; + self.follow_seek_head(io, file, &[INFO, TRACKS], u64::MAX)?; if !self.opts.index { break; } @@ -821,22 +926,42 @@ impl Walker<'_> { if io.cache.recovery_left == 0 { warn(&mut file.warnings, "recovery budget spent; the rest of the damaged data was not searched".into()); } - self.follow_seek_head(io, file, &[INFO, TRACKS, CUES, CHAPTERS, TAGS, ATTACHMENTS, SEEK_HEAD])?; + if track { + self.save(file, pos); + } + if stopped { + // what lies after the scanned part (Cues, Tags…) waits for the rest of the index + let before = file.index_end.unwrap_or(pos); + self.follow_seek_head(io, file, &[INFO, TRACKS, CUES, CHAPTERS, TAGS, ATTACHMENTS, SEEK_HEAD], before)?; + file.cues.sort_by_key(|c| c.time); + file.indexed = false; + return Ok(()); + } + self.follow_seek_head(io, file, &[INFO, TRACKS, CUES, CHAPTERS, TAGS, ATTACHMENTS, SEEK_HEAD], u64::MAX)?; // follow a second-level SeekHead reached through the first one - self.follow_seek_head(io, file, &[CUES, CHAPTERS, TAGS, ATTACHMENTS])?; + self.follow_seek_head(io, file, &[CUES, CHAPTERS, TAGS, ATTACHMENTS], u64::MAX)?; file.cues.sort_by_key(|c| c.time); file.indexed = self.opts.index; + file.index_end = None; + file.resume = None; Ok(()) } - fn follow_seek_head(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, ids: &[u32]) -> Result<()> { + /// Record that everything before `pos` is done: a read that fails from here on (data not + /// loaded yet) can be retried by [`MkvFile::extend_index`] without repeating any of it. + fn save(&self, file: &mut MkvFile, pos: u64) { + file.index_end = Some(pos); + } + + /// Read the elements of the given kinds that the SeekHead points at below `before`. + fn follow_seek_head(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, ids: &[u32], before: u64) -> Result<()> { let entries = file.seek_head.clone(); for e in entries { if !ids.contains(&e.id) { continue; } let Some(abs) = file.segment_data_start.checked_add(e.position) else { continue }; - if self.parsed.contains_key(&abs) || abs >= file.segment_end { + if self.parsed.contains_key(&abs) || abs >= file.segment_end || abs >= before { continue; } match io.header(abs, file.segment_end)? { @@ -859,8 +984,21 @@ impl Walker<'_> { } } - /// Parse a level-1 metadata element; returns its end. + /// Parse a level-1 metadata element; returns its end. An element whose read fails is left + /// unrecorded, so a retry reads it again and one that succeeded is never read twice. fn level1(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, h: &Hdr, limit: u64) -> Result { + let known = self.parsed.contains_key(&h.pos); + let (meta_left, attachments) = (self.meta_left, file.attachments.len()); + let result = self.level1_read(io, file, h, limit); + if result.is_err() && !known { + self.parsed.remove(&h.pos); + self.meta_left = meta_left; + file.attachments.truncate(attachments); + } + result + } + + fn level1_read(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, h: &Hdr, limit: u64) -> Result { if let Some(&end) = self.parsed.get(&h.pos) { return Ok(end); } @@ -999,6 +1137,27 @@ impl MkvFile { Some((end.max(0) as u64).saturating_mul(self.info.timestamp_scale)) } + /// Continue a scan that an [`OpenOptions::index_limit`] stopped, from [`MkvFile::index_end`], + /// without rescanning what is indexed. `opts.index_limit` bounds the new scan like at open + /// (`None`: to the end of the Segment, which completes the index: Cues, Tags… are read and + /// [`MkvFile::indexed`] becomes `true`); `opts.verify_crc` applies as at open. + /// + /// A no-op when there is nothing left to scan, including a limit at or below `index_end`. + /// Safe to call again after an error such as a byte source that has not loaded the range yet + /// (`WouldBlock`): the clusters scanned before the failure stay indexed and `index_end` says + /// where the next call continues. + pub fn extend_index(&mut self, src: &S, opts: &OpenOptions) -> Result<()> { + let Some(start) = self.index_end else { return Ok(()) }; + let resume = self.resume.take().unwrap_or_default(); + let opts = OpenOptions { index: true, ..opts.clone() }; + let mut cache = Cache::default(); + let mut io = Io::new(src, &mut cache); + let budget = FrameBudget::resumed(io.len, &self.tracks); + let mut w = + Walker { opts: &opts, parsed: resume.parsed, meta_left: resume.meta_left, blocks_seen: true, resuming: true, lookup: None, budget }; + w.walk_from(&mut io, self, start) + } + /// Index of the track with this `TrackNumber`. pub fn track_by_number(&self, number: u64) -> Option { self.tracks.iter().position(|t| t.number == number) @@ -1117,12 +1276,16 @@ impl Demuxer { self.file.duration_ns() } - /// Scan all clusters and build sample tables (no-op if already indexed). + /// Scan all clusters and build sample tables (no-op if already indexed; continues a scan that + /// stopped at [`OpenOptions::index_limit`] from where it stopped). pub fn build_index(&mut self) -> Result<()> { if self.file.indexed { return Ok(()); } - let opts = OpenOptions { index: true, verify_crc: self.opts.verify_crc }; + let opts = OpenOptions { index: true, verify_crc: self.opts.verify_crc, index_limit: None }; + if self.file.index_end.is_some() { + return self.file.extend_index(&self.src, &opts); + } self.file = open_impl(&self.src, &opts, &mut self.cache)?; // the index pass spent its own budget; packet reading keeps a whole one self.cache.recovery_left = RECOVERY_BUDGET; diff --git a/crates/media/src/matroska/index_limit_tests.rs b/crates/media/src/matroska/index_limit_tests.rs new file mode 100644 index 000000000..059a1388b --- /dev/null +++ b/crates/media/src/matroska/index_limit_tests.rs @@ -0,0 +1,269 @@ +//! Limited indexing ([`OpenOptions::index_limit`]) and [`MkvFile::extend_index`], on files written +//! with [`MkvWriter`] (one Cluster per second, Cues at the end). + +use std::cell::Cell; +use std::io::{Cursor, ErrorKind}; + +use crate::matroska::{ByteSource, Demuxer, MkvFile, MkvWriter, MuxOptions, OpenOptions, TrackKind, TrackSpec, open_with}; + +/// A video track of `frames` 100 ms frames (every fifth a keyframe): ten frames per Cluster. +fn file(frames: i64) -> Vec { + let mut v = TrackSpec::new(TrackKind::Video, "V_MJPEG"); + v.video_size = Some((16, 16)); + v.default_duration_ns = Some(100_000_000); + let mut w = MkvWriter::new(Cursor::new(Vec::new()), vec![v], MuxOptions::default()).expect("writer"); + for k in 0..frames { + w.write_frame(0, k * 100_000_000, k % 5 == 0, &[k as u8; 200], Some(100_000_000)).expect("frame"); + } + w.finish().expect("finish").into_inner() +} + +fn limited(bytes: &[u8], limit: Option) -> MkvFile { + open_with(bytes, &OpenOptions { index_limit: limit, ..OpenOptions::default() }).expect("open") +} + +fn unlimited() -> OpenOptions { + OpenOptions::default() +} + +/// The sample tables, cluster list and cues of two files are equal. +fn assert_same(a: &MkvFile, b: &MkvFile) { + assert_eq!(a.tracks.len(), b.tracks.len()); + for (x, y) in a.tracks.iter().zip(&b.tracks) { + assert_eq!(x.samples, y.samples); + assert_eq!(x.sync, y.sync); + } + assert_eq!(a.clusters, b.clusters); + assert_eq!(a.cues, b.cues); + assert_eq!((a.indexed, a.index_end), (b.indexed, b.index_end)); +} + +#[test] +fn default_options_are_unchanged() { + let bytes = file(45); + let f = open_with(&bytes[..], &OpenOptions::default()).expect("open"); + assert!(f.indexed && f.index_end.is_none()); + assert_eq!(f.clusters.len(), 5); + assert_eq!(f.tracks[0].samples.len(), 45); + assert!(!f.cues.is_empty()); +} + +#[test] +fn limit_stops_before_a_cluster_that_does_not_fit() { + let bytes = file(45); + let full = limited(&bytes, None); + let c = &full.clusters; + // the limit lies in the middle of the third cluster: two are indexed, the third waits + let f = limited(&bytes, Some(c[2].offset + 40)); + assert_eq!(f.clusters, c[..2]); + assert_eq!(f.tracks[0].samples, full.tracks[0].samples[..20]); + assert!(!f.indexed); + assert_eq!(f.index_end, Some(c[2].offset)); + // Info and Tracks are there; the Cues at the end are not read yet + assert_eq!(f.info.duration, full.info.duration); + assert_eq!(f.duration_ns(), full.duration_ns()); + assert!(f.cues.is_empty()); + // a limit exactly at the end of a cluster keeps that cluster + let g = limited(&bytes, Some(c[1].end)); + assert_eq!(g.clusters.len(), 2); + assert_eq!(g.index_end, Some(c[1].end)); +} + +#[test] +fn limit_before_the_first_cluster_still_scans_it() { + let bytes = file(45); + let full = limited(&bytes, None); + for limit in [0, 1, full.first_cluster.expect("cluster") / 2] { + let f = limited(&bytes, Some(limit)); + assert_eq!(f.clusters.len(), 1, "limit {limit}"); + assert_eq!(f.tracks[0].samples.len(), 10); + assert_eq!(f.index_end, Some(full.clusters[0].end)); + assert!(!f.indexed); + } +} + +#[test] +fn limit_past_the_end_is_a_full_index() { + let bytes = file(45); + let full = limited(&bytes, None); + for limit in [bytes.len() as u64, bytes.len() as u64 + 1, u64::MAX] { + assert_same(&limited(&bytes, Some(limit)), &full); + } +} + +#[test] +fn limit_without_an_index_is_ignored() { + let bytes = file(45); + let f = open_with(&bytes[..], &OpenOptions { index: false, verify_crc: false, index_limit: Some(0) }).expect("open"); + assert!(!f.indexed && f.index_end.is_none() && f.clusters.is_empty()); + assert!(!f.cues.is_empty(), "the SeekHead still leads to the Cues"); +} + +#[test] +fn extending_in_steps_equals_a_full_open() { + let bytes = file(45); + let full = limited(&bytes, None); + let mut f = limited(&bytes, Some(0)); + let mut seen = vec![f.clusters.len()]; + for c in &full.clusters { + f.extend_index(&bytes[..], &OpenOptions { index_limit: Some(c.end), ..unlimited() }).expect("extend"); + seen.push(f.clusters.len()); + } + assert_eq!(seen, [1, 1, 2, 3, 4, 5]); + // the Cues after the last cluster wait for the rest of the index + assert!(!f.indexed && f.index_end.is_some() && f.cues.is_empty()); + f.extend_index(&bytes[..], &unlimited()).expect("complete"); + assert_same(&f, &full); +} + +#[test] +fn extending_twice_with_the_same_limit_changes_nothing() { + let bytes = file(45); + let mut f = limited(&bytes, Some(0)); + let limit = f.clusters[0].end + 5_000; + let opts = OpenOptions { index_limit: Some(limit), ..unlimited() }; + f.extend_index(&bytes[..], &opts).expect("extend"); + let (samples, end) = (f.tracks[0].samples.clone(), f.index_end); + f.extend_index(&bytes[..], &opts).expect("extend again"); + assert_eq!((f.tracks[0].samples.clone(), f.index_end), (samples, end)); + // a lower limit than what is indexed is a no-op too + f.extend_index(&bytes[..], &OpenOptions { index_limit: Some(0), ..unlimited() }).expect("extend lower"); + assert_eq!(f.index_end, end); + // to the end, then once more on a complete index + f.extend_index(&bytes[..], &unlimited()).expect("complete"); + assert!(f.indexed); + f.extend_index(&bytes[..], &unlimited()).expect("complete again"); + assert_same(&f, &limited(&bytes, None)); +} + +/// A source that has loaded only a prefix: reads beyond it fail like the web build's pending reads. +struct Loaded<'a> { + data: &'a [u8], + loaded: Cell, +} + +impl ByteSource for Loaded<'_> { + fn len(&self) -> u64 { + self.data.len() as u64 + } + fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<()> { + if offset + buf.len() as u64 > self.loaded.get() { + return Err(std::io::Error::new(ErrorKind::WouldBlock, "not loaded yet")); + } + self.data.read_at(offset, buf) + } +} + +#[test] +fn a_growing_prefix_indexes_without_reading_ahead() { + let bytes = file(45); + let full = limited(&bytes, None); + let src = Loaded { data: &bytes, loaded: Cell::new(0) }; + // nothing loaded: the open fails with WouldBlock instead of panicking + let opts = |limit: u64| OpenOptions { index_limit: Some(limit), ..unlimited() }; + let err = open_with(&src, &opts(0)).expect_err("nothing loaded"); + assert!(err.to_string().contains("not loaded"), "{err}"); + // the head up to the end of the first cluster is enough to open, however much more is asked for + src.loaded.set(full.clusters[0].end); + let mut f = open_with(&src, &opts(src.loaded.get())).expect("first cluster loaded"); + assert_eq!(f.clusters.len(), 1); + // each chunk that arrives indexes the clusters it holds completely, and nothing past them + for (n, c) in full.clusters.iter().enumerate().skip(1) { + src.loaded.set(c.end - 1); + f.extend_index(&src, &opts(src.loaded.get())).expect("extend"); + assert_eq!(f.clusters.len(), n, "a cluster missing its last byte waits"); + src.loaded.set(c.end); + f.extend_index(&src, &opts(src.loaded.get())).expect("extend"); + assert_eq!(f.clusters.len(), n + 1); + } + // the tail (Cues) arrives: the index completes + src.loaded.set(bytes.len() as u64); + f.extend_index(&src, &unlimited()).expect("complete"); + assert_same(&f, &full); +} + +/// A source whose read number `fail_at` fails once with `WouldBlock`, whatever it asks for. +struct Flaky<'a> { + data: &'a [u8], + calls: Cell, + fail_at: usize, +} + +impl ByteSource for Flaky<'_> { + fn len(&self) -> u64 { + self.data.len() as u64 + } + fn read_at(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<()> { + self.calls.set(self.calls.get() + 1); + if self.calls.get() == self.fail_at { + return Err(std::io::Error::new(ErrorKind::WouldBlock, "not loaded yet")); + } + self.data.read_at(offset, buf) + } +} + +#[test] +fn a_failed_extension_resumes_without_duplicates() { + // Tags between the clusters and Cues after them make the elements read twice visible + let bytes = file(45); + let full = limited(&bytes, None); + let mut failed = 0; + for fail_at in 1..40 { + let mut f = limited(&bytes, Some(0)); + let src = Flaky { data: &bytes, calls: Cell::new(0), fail_at }; + // a read that fails anywhere in the scan leaves the file consistent: the retry finishes it + if f.extend_index(&src, &unlimited()).is_err() { + failed += 1; + assert!(f.index_end.is_some() && !f.indexed, "fail_at {fail_at}"); + assert_eq!(f.clusters.len(), f.tracks[0].samples.len() / 10, "whole clusters only, fail_at {fail_at}"); + f.extend_index(&bytes[..], &unlimited()).expect("resume"); + } + assert_same(&f, &full); + } + assert!(failed > 0, "the failing read was reached"); +} + +#[test] +fn truncated_files_never_panic() { + let bytes = file(25); + for len in 0..bytes.len() { + for limit in [Some(0), Some(len as u64 / 2), None] { + let cut = &bytes[..len]; + let Ok(mut f) = open_with(cut, &OpenOptions { index_limit: limit, ..unlimited() }) else { continue }; + let _ = f.extend_index(cut, &unlimited()); + let _ = f.extend_index(cut, &OpenOptions { index_limit: Some(1), ..unlimited() }); + assert!(f.index_end.is_none() || !f.indexed); + } + } +} + +#[test] +fn demuxer_completes_a_limited_index_and_seeks() { + let bytes = file(45); + let full = limited(&bytes, None); + let mut d = Demuxer::with_options(&bytes[..], OpenOptions { index_limit: Some(0), ..unlimited() }).expect("open"); + assert_eq!(d.file().clusters.len(), 1); + // packets stream from the first Cluster on, whatever the index covers + assert_eq!(d.next_packet().expect("packet").expect("some").pts, 0); + let keys = d.keyframe_index(0).expect("keyframes"); + assert!(d.file().indexed); + assert_eq!(keys.len(), full.tracks[0].sync.len()); + let at = d.seek(0, 3_500_000_000).expect("seek"); + assert_eq!(at.pts_ns, 3_500_000_000); +} + +#[test] +fn a_tiny_last_block_needs_no_bytes_past_its_cluster() { + let mut v = TrackSpec::new(TrackKind::Video, "V_MJPEG"); + v.video_size = Some((16, 16)); + let mut w = MkvWriter::new(Cursor::new(Vec::new()), vec![v], MuxOptions::default()).expect("writer"); + for k in 0..30 { + w.write_frame(0, k * 100_000_000, true, &[k as u8], Some(100_000_000)).expect("frame"); + } + let bytes = w.finish().expect("finish").into_inner(); + let full = limited(&bytes, None); + // a 7-byte block ends each cluster: its header read must not reach into the next cluster + let src = Loaded { data: &bytes, loaded: Cell::new(full.clusters[0].end) }; + let f = open_with(&src, &OpenOptions { index_limit: Some(src.loaded.get()), ..unlimited() }).expect("open"); + assert_eq!(f.clusters.len(), 1); +} diff --git a/crates/media/src/matroska/mod.rs b/crates/media/src/matroska/mod.rs index c0d7dae2b..9e60a9600 100644 --- a/crates/media/src/matroska/mod.rs +++ b/crates/media/src/matroska/mod.rs @@ -15,6 +15,8 @@ mod demux; pub mod ebml; mod error; mod ids; +#[cfg(test)] +mod index_limit_tests; mod meta; mod mux; #[cfg(test)] diff --git a/crates/media/src/matroska/synthetic_tests.rs b/crates/media/src/matroska/synthetic_tests.rs index d1da05285..d667a0952 100644 --- a/crates/media/src/matroska/synthetic_tests.rs +++ b/crates/media/src/matroska/synthetic_tests.rs @@ -203,12 +203,12 @@ fn assert_crc_mismatch_is_reported(bytes: &[u8]) { let mut corrupted = bytes.to_vec(); let title = corrupted.windows(9).position(|window| window == b"Synthetic").unwrap(); corrupted[title] = b's'; - let file = open_with(&corrupted[..], &OpenOptions { index: true, verify_crc: true }).unwrap(); + let file = open_with(&corrupted[..], &OpenOptions { index: true, verify_crc: true, index_limit: None }).unwrap(); assert_eq!(file.crc_errors.len(), 1); } fn assert_seek_builds_an_index(bytes: &[u8]) { - let mut demuxer = Demuxer::with_options(bytes, OpenOptions { index: false, verify_crc: false }).unwrap(); + let mut demuxer = Demuxer::with_options(bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }).unwrap(); assert_eq!(demuxer.file().tracks.len(), 2); assert_eq!(demuxer.seek(0, 1_100_000_000).unwrap().pts, 1000); assert_eq!(demuxer.seek(0, 1_130_000_000).unwrap().pts, 1120); @@ -219,7 +219,7 @@ fn assert_seek_builds_an_index(bytes: &[u8]) { #[test] fn synthetic_file() { let bytes = synthetic(); - let file = open_with(&bytes[..], &OpenOptions { index: true, verify_crc: true }).unwrap(); + let file = open_with(&bytes[..], &OpenOptions { index: true, verify_crc: true, index_limit: None }).unwrap(); assert_synthetic_metadata(&file, &bytes); assert_synthetic_packets(&file, &bytes); assert_crc_mismatch_is_reported(&bytes); @@ -327,7 +327,7 @@ fn cue_position_past_the_file_falls_back_to_the_index() { let mut el_cues = Vec::new(); el(&mut el_cues, CUES, &cues); bytes.splice(at..at, el_cues); - let mut d = Demuxer::with_options(bytes, OpenOptions { index: false, verify_crc: false }).unwrap(); + let mut d = Demuxer::with_options(bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }).unwrap(); assert_eq!(d.file().cues.len(), 1); assert_eq!(d.seek(0, 0).unwrap().pts, 1000); } @@ -449,7 +449,7 @@ fn overlapping_seek_head_targets_are_read_once() { } f.resize(f.len() + pad, 0xEC); let src = Counting(&f, std::cell::Cell::new(0)); - let file = open_with(&src, &OpenOptions { index: false, verify_crc: false }).unwrap(); + let file = open_with(&src, &OpenOptions { index: false, verify_crc: false, index_limit: None }).unwrap(); assert_eq!(file.seek_head.len(), n as usize); assert!(src.1.get() <= 4 * f.len() as u64, "read {} bytes of a {}-byte file", src.1.get(), f.len()); } @@ -609,7 +609,7 @@ fn damaged_before_a_cluster() -> Vec { #[test] fn building_the_index_gets_its_own_recovery_budget() { let f = damaged_before_a_cluster(); - let opts = OpenOptions { index: false, verify_crc: false }; + let opts = OpenOptions { index: false, verify_crc: false, index_limit: None }; let one = open_with(&f[..], &OpenOptions::default()).unwrap(); assert_eq!(one.tracks[0].samples.len(), 1, "one pass finds the cluster: {:?}", one.warnings); let mut d = Demuxer::with_options(&f[..], opts).unwrap(); @@ -634,7 +634,7 @@ fn packet_reading_keeps_its_recovery_budget_after_the_index_is_built() { write_unknown_size(&mut f, 8); el_uint(&mut f, TIMESTAMP, 10); el(&mut f, SIMPLE_BLOCK, &block(1, 0, 0x80, &[2])); - let mut d = Demuxer::with_options(&f[..], OpenOptions { index: false, verify_crc: false }).unwrap(); + let mut d = Demuxer::with_options(&f[..], OpenOptions { index: false, verify_crc: false, index_limit: None }).unwrap(); d.build_index().unwrap(); assert_eq!(d.file().tracks[0].samples.len(), 2, "the index found both clusters: {:?}", d.file().warnings); let mut n = 0; diff --git a/crates/media/tests/fuzz_matroska.rs b/crates/media/tests/fuzz_matroska.rs index 73a433616..e324ad61e 100644 --- a/crates/media/tests/fuzz_matroska.rs +++ b/crates/media/tests/fuzz_matroska.rs @@ -47,7 +47,8 @@ fn synthetic_files() -> Vec> { fn exercise(data: &[u8]) -> bool { let mut parsed = false; - for opts in [OpenOptions { index: true, verify_crc: true }, OpenOptions { index: false, verify_crc: false }] { + for opts in [OpenOptions { index: true, verify_crc: true, index_limit: None }, OpenOptions { index: false, verify_crc: false, index_limit: None }] + { let Ok(f) = open_with(data, &opts) else { continue }; parsed = true; let _ = f.duration_ns(); diff --git a/crates/media/tests/oracle_matroska.rs b/crates/media/tests/oracle_matroska.rs index c7476aaaf..d4da588e7 100644 --- a/crates/media/tests/oracle_matroska.rs +++ b/crates/media/tests/oracle_matroska.rs @@ -168,12 +168,12 @@ fn h264_aac() { assert!(!f.cues.is_empty()); assert!(!f.seek_head.is_empty()); check_seeks(&bytes, OpenOptions::default(), v, &packets); - check_seeks(&bytes, OpenOptions { index: false, verify_crc: false }, v, &packets); + check_seeks(&bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }, v, &packets); check_seeks(&bytes, OpenOptions::default(), a, &packets); // Read + Seek entry point and CRC verification (ffmpeg writes CRC-32 in level-1 elements) let d = Demuxer::from_reader(std::io::Cursor::new(bytes.clone())).unwrap(); assert_eq!(d.file().tracks.len(), 2); - let crc = open_with(&bytes[..], &OpenOptions { index: true, verify_crc: true }).unwrap(); + let crc = open_with(&bytes[..], &OpenOptions { index: true, verify_crc: true, index_limit: None }).unwrap(); assert!(crc.crc_checked > 0); assert!(crc.crc_errors.is_empty()); let kf = Demuxer::from_slice(&bytes).unwrap().keyframe_index(v).unwrap(); @@ -192,7 +192,7 @@ fn vp9_opus_webm() { let v = f.track_of_kind(TrackKind::Video).unwrap(); assert!(matches!(f.tracks[v].codec, Codec::Vp9 { .. })); check_seeks(&bytes, OpenOptions::default(), v, &packets); - check_seeks(&bytes, OpenOptions { index: false, verify_crc: false }, v, &packets); + check_seeks(&bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }, v, &packets); } #[test] @@ -241,7 +241,7 @@ fn cues_at_front() { assert!(!f.cues.is_empty()); assert!(f.clusters.len() >= 5); let v = f.track_of_kind(TrackKind::Video).unwrap(); - check_seeks(&bytes, OpenOptions { index: false, verify_crc: false }, v, &packets); + check_seeks(&bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }, v, &packets); } #[test] @@ -252,8 +252,8 @@ fn live_stream_without_cues() { assert!(f.duration_ns().unwrap() > 1_900_000_000); let v = f.track_of_kind(TrackKind::Video).unwrap(); // no cues: seeking builds the index by scanning clusters - check_seeks(&bytes, OpenOptions { index: false, verify_crc: false }, v, &packets); - let mut d = Demuxer::with_options(&bytes[..], OpenOptions { index: false, verify_crc: false }).unwrap(); + check_seeks(&bytes, OpenOptions { index: false, verify_crc: false, index_limit: None }, v, &packets); + let mut d = Demuxer::with_options(&bytes[..], OpenOptions { index: false, verify_crc: false, index_limit: None }).unwrap(); assert!(!d.file().indexed); assert!(!d.keyframe_index(v).unwrap().is_empty()); assert!(d.file().indexed);