craft-media: index a Matroska file progressively (#430)

* craft-media: index a Matroska file progressively (OpenOptions::index_limit, MkvFile::extend_index)

* craft-media: Matroska element headers never read past their parent

---------

Co-authored-by: Dylan M. Taylor <dylan@dylanmtaylor.com>

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

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::<u64>();
@@ -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<u64>,
 }
 
 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<u64>,
     /// 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<u64>,
+    /// Walker state to continue the scan from [`MkvFile::index_end`].
+    resume: Option<Resume>,
     /// 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<String>,
 }
 
+/// 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<u64, u64>,
+    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<S: ByteSource + ?Sized>(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<S: ByteSource + ?Sized>(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<S: ByteSource + ?Sized>(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<TrackIndex>,
     budget: FrameBudget,
@@ -763,9 +827,42 @@ struct Walker<'o> {
 
 impl Walker<'_> {
     fn walk<S: ByteSource + ?Sized>(&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<S: ByteSource + ?Sized>(&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<S: ByteSource + ?Sized>(&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<S: ByteSource + ?Sized>(&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<S: ByteSource + ?Sized>(&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<S: ByteSource + ?Sized>(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, h: &Hdr, limit: u64) -> Result<u64> {
+        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<S: ByteSource + ?Sized>(&mut self, io: &mut Io<'_, S>, file: &mut MkvFile, h: &Hdr, limit: u64) -> Result<u64> {
         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<S: ByteSource + ?Sized>(&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<usize> {
         self.tracks.iter().position(|t| t.number == number)
@@ -1117,12 +1276,16 @@ impl<S: ByteSource> Demuxer<S> {
         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;

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<u8> {
+    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<u64>) -> 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<u64>,
+}
+
+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<usize>,
+    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);
+}

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)]

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<u8> {
 #[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;

crates/media/tests/fuzz_matroska.rs

@@ -47,7 +47,8 @@ fn synthetic_files() -> Vec<Vec<u8>> {
 
 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();

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);