diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index fa838cbc3c6..1ba340321d0 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -13,7 +13,7 @@ back. This is what `moq import` / `export` and the gateways are built on. | Format | Import | Export | Notes | | --- | --- | --- | --- | -| fMP4 / CMAF | yes | yes | Audio/video unwrapped to `legacy` by default, or passed through as `cmaf` with `with_passthrough`; `wvtt` subtitles import as timed WebVTT text and `stpp` TTML (IMSC timing profile) as `ttml` documents on the broadcast clock, preserving source language and label when present; other subtitle codecs are skipped. | +| fMP4 / CMAF | yes | yes | Audio/video unwrapped to `legacy` by default, or passed through as `cmaf` with `with_passthrough`; `wvtt` subtitles import as timed WebVTT text and `stpp` TTML (IMSC timing profile) as `ttml` documents on the broadcast clock, preserving source language and label when present; other subtitle codecs are skipped. Import streams media samples as they arrive; each track's samples must be in forward byte order, and a `moov`, `moof`, or text sample over 32 MiB is refused. | | MPEG-TS | yes | yes | H.264/H.265; AAC, MP2, AC-3, E-AC-3, Opus up to 7.1; SCTE-35 sections and every other undecoded PES stream (KLV, subtitles, DVB AC-3) as tracks; service tables round-trip. | | FLV / RTMP | yes | yes | Legacy H.264 + AAC + MP3, plus enhanced-RTMP HEVC, AV1, VP9, Opus, AC-3, E-AC-3, and multitrack. | | Matroska / WebM | yes | yes | Import streams each block into its frame in one pass, retaining no declared size. It refuses subtitle tracks, CMAF output, and video `BlockGroup`s with fields after their `Block`: remux video as `SimpleBlock`s and extract subtitles separately. | diff --git a/quest/m1/security-audit/README.md b/quest/m1/security-audit/README.md index 4395dd15073..3e68e4393ca 100644 --- a/quest/m1/security-audit/README.md +++ b/quest/m1/security-audit/README.md @@ -54,6 +54,5 @@ Dropped, with reasons: ## Required -- [fMP4 import streams mdat](/quest/m1/security-audit/fmp4-stream-mdat.md) - a box header declaring a huge size cannot pin importer memory across segments - [Browser input checks](/quest/m1/security-audit/js-input.md) - room metadata is schema-validated and the http:// fingerprint fetch is aborted and length-checked - [CI tool pins](/quest/m1/security-audit/ci-pins.md) - binding generators are pinned by commit in one place, and the remaining unpinned installs and broad token grants are fixed diff --git a/quest/m1/security-audit/fmp4-stream-mdat.md b/quest/m1/security-audit/fmp4-stream-mdat.md deleted file mode 100644 index 01513353ec9..00000000000 --- a/quest/m1/security-audit/fmp4-stream-mdat.md +++ /dev/null @@ -1,42 +0,0 @@ -# [M] fMP4 import streams mdat - -## Goal - -The fMP4 importer's memory follows its output's backpressure, not the sizes -an input declares. A box header claiming 2^62 bytes, or an mdat that never -completes, cannot grow the importer's buffer without bound, including across -HLS segments. - -## Plan - -From the 2026-10-27 audit (MUX-2), verified worse than reported: -`rs/moq-mux/src/container/fmp4/import.rs` appends input to `self.buffer` and -waits for `mp4_atom::Any::decode_maybe` to see a whole box. A declared size -past what is buffered keeps returning `None`, and the HLS importer carries a -trailing partial atom into the next segment, so one bogus header accumulates -every later segment. - -Decided 2026-10-27 (maintainer): no byte cap. Stream each mdat sample into -its frame as bytes arrive, resolving sample ranges from the parsed `moof` as -today, and discard skipped boxes (`free`, `skip`, unknown types) as their -bytes arrive instead of buffering them. This fits -[unbounded groups](/quest/m1/unbounded-groups/README.md), which removes the -frame cap so a sample can be large. Since -[8d2fa6653a](https://github.com/moq-dev/moq/commit/8d2fa6653a) a fragment can -hold several GOPs, so a fragment-sized cap would also refuse legal input. - -`moov` and `moof` are parsed whole today, so a header declaring a huge one -still pins input. Decided 2026-10-28 (maintainer, for both importers): parse -metadata children incrementally, keeping only the fields consumed, with no -blanket byte cap. MKV import does so since -[moq-dev/moq#7234](https://github.com/moq-dev/moq/pull/7234). - -Keep `sample::Resolver`'s range checks and `Claimed`'s no-reuse rule. Tests -feed a huge declared `moof`, `moov`, `mdat`, and skipped box in small chunks -and assert the importer's held bytes stay bounded. - -Public API: none expected. Wire: none. - -## Related - -- [Media fuzz](/quest/m2/media-fuzz.md) - would catch this class of buffer lifetime bug diff --git a/rs/moq-mux/benches/containers.rs b/rs/moq-mux/benches/containers.rs index 21616ef220c..b45d80f4b2f 100644 --- a/rs/moq-mux/benches/containers.rs +++ b/rs/moq-mux/benches/containers.rs @@ -132,10 +132,20 @@ fn cmaf_import(passthrough: bool) -> (moq_mux::catalog::Producer, moq_mux::conta } fn import_cmaf( + import: (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import), + data: &Bytes, +) -> (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import) { + import_cmaf_chunks(import, data, data.len()) +} + +fn import_cmaf_chunks( (catalog, mut import): (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import), data: &Bytes, + chunk: usize, ) -> (moq_mux::catalog::Producer, moq_mux::container::fmp4::Import) { - import.decode(data).unwrap(); + for chunk in data.chunks(chunk) { + import.decode(chunk).unwrap(); + } import.finish().unwrap(); (catalog, import) } @@ -272,16 +282,22 @@ fn interleaved_runs(c: &mut Criterion) { for (direction, passthrough) in [("import", true), ("unwrap", false)] { let mut group = c.benchmark_group(format!("cmaf-interleaved-{direction}")); group.throughput(Throughput::Bytes(data.len() as u64)); - group.bench_function( - BenchmarkId::new(format!("{tracks}tracks"), format!("{runs}runs-{size}bytes")), - |b| { - b.iter_batched( - || cmaf_import(passthrough), - |import| import_cmaf(import, std::hint::black_box(&data)), - BatchSize::PerIteration, - ); - }, - ); + for chunk in [1, 1024, data.len()] { + group.bench_function( + BenchmarkId::new( + format!("{tracks}tracks"), + format!("{runs}runs-{size}bytes-{chunk}chunk"), + ), + |b| { + b.iter_batched( + || cmaf_import(passthrough), + |import| import_cmaf_chunks(import, std::hint::black_box(&data), chunk), + BatchSize::PerIteration, + ); + }, + ); + } + group.finish(); } } diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 211bc0705ca..ee3cf7603b9 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -1,12 +1,13 @@ use bytes::{Bytes, BytesMut}; use hang::catalog::{AAC, AudioCodec, AudioConfig, Container, H264, H265, VP9, VideoCodec, VideoConfig}; use moq_net::Timestamp; -use mp4_atom::{Any, Atom, DecodeMaybe, Encode, Mdat, Moof, Moov, Trak}; +use mp4_atom::{Atom, DecodeMaybe, Encode, Moof, Moov, Trak}; use std::collections::{HashMap, HashSet}; use super::{Error, Kind, edit::Edit, sample}; use crate::Result; use crate::catalog::Estimator; +use crate::container::producer::Streamed; /// Imports fragmented MP4 (fMP4/CMAF) into a MoQ broadcast. /// @@ -75,6 +76,7 @@ pub struct Import { // Bytes carried across calls: a partial atom at the tail of one `decode` waits // here for the rest to arrive on the next call. buffer: BytesMut, + body: Option, // A segment boundary (a styp or an explicit `cut()`) waiting to land on the next fragment. pending_cut: bool, @@ -87,6 +89,9 @@ pub struct Import { // keyframe), so each track cuts its HLS segments where it opens its group for the segment. declared: bool, dropped_metadata: u64, + + // A seek or discontinuity waiting for the fragment still publishing. + deferred: Deferred, } /// The catalog entry for one passthrough track, whichever section it lives in. @@ -270,10 +275,12 @@ impl Import { moof_span: 0, broadcast, buffer: BytesMut::new(), + body: None, pending_cut: false, segment: 0, declared: false, dropped_metadata: 0, + deferred: Deferred::default(), } } @@ -332,75 +339,189 @@ impl Import { /// Decode a buffer of bytes. /// - /// Samples must be nonempty and use moof-relative offsets; absolute `base_data_offset` - /// addressing is unsupported. With ffmpeg, use `default_base_moof`. - pub fn decode(&mut self, data: &[u8]) -> Result<()> { - self.buffer.extend_from_slice(data); - self.drain() - } - - /// Parse every whole top-level atom buffered so far, leaving any trailing - /// partial atom for the next call. - fn drain(&mut self) -> Result<()> { - // Parse complete atoms first, recording each one's byte range, then process - // them. Collecting up front keeps `self.buffer` un-borrowed while the handlers - // (`init`/`extract`) take `&mut self`. - let mut parsed = Vec::new(); - let mut position = 0; + /// Samples must be nonempty, have forwards per-track byte ranges, and use moof-relative + /// offsets; absolute `base_data_offset` addressing is unsupported. With ffmpeg, use + /// `default_base_moof`. Forward gaps and interleaved tracks are supported. + /// + /// Media samples stream into their frames as bytes arrive. What must be held whole is + /// bounded: a `moov` or `moof` body, or a text sample, larger than + /// [`MAX_HELD_BYTES`](super::MAX_HELD_BYTES), or a `moof` declaring more than + /// [`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<()> { loop { - let mut cursor = std::io::Cursor::new(&self.buffer[position..]); - let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor)? else { - break; - }; - let size = cursor.position() as usize; - parsed.push((atom, position, size)); - position += size; - } - - if position == 0 { - return Ok(()); - } - - // Detach the fully-parsed prefix as a cheap ref-counted buffer so each mdat's - // raw bytes can be sliced out without copying or borrowing `self`. - let consumed = self.buffer.split_to(position).freeze(); - - for (atom, start, size) in parsed { - if self.moof.is_some() { - self.moof_span = self.moof_span.checked_add(size).ok_or(Error::InvalidDataOffset)?; - } - match atom { - Any::Ftyp(_) => {} - Any::Emsg(_) => { - self.dropped_metadata += 1; - if self.dropped_metadata == 1 { - tracing::warn!(count = self.dropped_metadata, "discarding fMP4 emsg metadata"); + let Some(body) = self.body.take() else { + let needed = if self.buffer.len() >= 4 && self.buffer[..4] == [0, 0, 0, 1] { + 16 + } else { + 8 + }; + let count = (needed - self.buffer.len()).min(data.len()); + self.buffer.extend_from_slice(&data[..count]); + data = &data[count..]; + let Some(header) = mp4_atom::Header::decode_maybe(&mut self.buffer.as_ref())? else { + if data.is_empty() { + return Ok(()); } + continue; + }; + let header_size = self.buffer.len(); + self.buffer.clear(); + if self.moof.is_some() { + self.moof_span = self + .moof_span + .checked_add(header_size) + .ok_or(Error::InvalidDataOffset)?; } - // A styp opens a CMAF segment: the on-disk segmentation is the boundary, - // landing on the next fragment. - Any::Styp(_) => self.pending_cut = true, - Any::Moov(moov) => { - self.init(moov)?; + let body = match header.kind.as_ref() { + b"moov" | b"moof" => self.begin_metadata(header.kind, header.size, header_size), + b"mdat" => self.begin_mdat(header.size).map(Body::Media), + kind => { + if kind == b"styp" { + self.pending_cut = true; + } + if kind == b"emsg" { + self.dropped_metadata += 1; + if self.dropped_metadata == 1 { + tracing::warn!(count = self.dropped_metadata, "discarding fMP4 emsg metadata"); + } + } + 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); + } } - Any::Moof(moof) => { + continue; + }; + match body { + Body::Skip(mut remaining) => { + let count = remaining.unwrap_or(data.len()).min(data.len()); + data = &data[count..]; if self.moof.is_some() { - return Err(Error::DuplicateMoof.into()); + self.moof_span = self.moof_span.checked_add(count).ok_or(Error::InvalidDataOffset)?; + } + if let Some(remaining) = &mut remaining { + *remaining -= count; + } + if remaining != Some(0) { + self.body = Some(Body::Skip(remaining)); } - self.moof.replace(moof); - self.moof_span = size; } - Any::Mdat(mdat) => { - let raw = consumed.slice(start..start + size); - self.extract(mdat, raw)?; + Body::Metadata { + kind, + mut remaining, + header_size, + } => { + let count = remaining.min(data.len()); + self.buffer.extend_from_slice(&data[..count]); + 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 { + self.body = Some(Body::Metadata { + kind, + remaining, + header_size, + }); + } } - _ => { - // Skip unknown atoms (e.g., sidx, which is optional and used for segment indexing) - // These are safe to ignore and don't affect playback + Body::Media(mut mdat) => { + let count = mdat.remaining.unwrap_or(data.len()).min(data.len()); + let written = self.write_mdat(&mut mdat, &data[..count]); + data = &data[count..]; + if let Some(remaining) = &mut mdat.remaining { + *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(); + if mdat.remaining != Some(0) { + self.body = Some(Body::Media(mdat)); + } + if published { + self.settle()?; + } } } + if data.is_empty() { + return Ok(()); + } } + } + + // Start holding a moov or moof body, refusing one too large to hold before buffering any of it. + fn begin_metadata(&mut self, kind: mp4_atom::FourCC, size: Option, header_size: usize) -> Result { + if kind == Moov::KIND && self.moov.is_some() { + return Err(Error::DuplicateMoov.into()); + } + if kind == Moof::KIND && self.moof.is_some() { + return Err(Error::DuplicateMoof.into()); + } + let size = size.ok_or(mp4_atom::Error::InvalidSize)?; + if size > super::MAX_HELD_BYTES { + return Err(Error::MetadataTooLarge { kind, size }.into()); + } + Ok(Body::Metadata { + kind, + remaining: size, + header_size, + }) + } + // Parse a held moov or moof body once its last byte has arrived. + fn end_metadata(&mut self, kind: mp4_atom::FourCC, body: Bytes, header_size: usize) -> Result<()> { + if kind == Moov::KIND { + return self.init(Moov::decode_body(&mut body.as_ref())?); + } + if self.moof.is_some() { + return Err(Error::DuplicateMoof.into()); + } + let moof = Moof::decode_body(&mut body.as_ref())?; + // Planning resolves every sample, and a run without per-sample fields declares its count + // in four bytes, so the byte bound alone does not limit what planning would hold. + let samples: u64 = moof + .traf + .iter() + .flat_map(|traf| &traf.trun) + .map(|trun| trun.samples.len() as u64) + .sum(); + if samples > super::MAX_FRAGMENT_SAMPLES { + return Err(Error::TooManySamples(samples).into()); + } + self.moof = Some(moof); + self.moof_span = header_size.checked_add(body.len()).ok_or(Error::InvalidDataOffset)?; Ok(()) } @@ -910,9 +1031,11 @@ impl Import { /// independently decodable, so it has no boundary of its own) follow the same segmentation as /// video instead of opening a group per fragment. /// - /// `mdat_len` bounds the scan across the whole moof: each sample takes at least one byte of the - /// mdat, and the resolver refuses any past it, so forged field-less runs of billions of samples, - /// however many trafs repeat them, aren't walked here. + /// The scan is bounded across the whole moof: `end_metadata` refuses a moof declaring more than + /// [`MAX_FRAGMENT_SAMPLES`](super::MAX_FRAGMENT_SAMPLES) samples before it becomes `self.moof`, + /// so forged field-less runs of billions of samples, however many trafs repeat them, aren't + /// walked here. `mdat_len` tightens that to one sample per mdat byte when the mdat declares its + /// size; a size-0 (to end of file) mdat passes `usize::MAX`. fn starts_segment(&self, moof: &Moof, has_video: bool, mdat_len: usize) -> bool { if self.pending_cut { return true; @@ -960,96 +1083,269 @@ impl Import { } // Publish the samples an mdat carries for each traf of the preceding moof. - fn extract(&mut self, mdat: Mdat, raw: Bytes) -> Result<()> { - let has_video = self.tracks.values().any(|t| t.kind == Kind::Video); + fn begin_mdat(&mut self, len: Option) -> Result { let moof = self.moof.take().ok_or(Error::NoMoof)?; - let header_size = raw.len() - mdat.data.len(); - // The mdat payload as a view of the input, so each unwrapped sample is sliced, not copied. - let payload = raw.slice(header_size..); - + if self.moov.is_none() { + return Err(Error::NoMoov.into()); + } // Resolve the segment boundary BEFORE the per-track loop. A moof can carry both an audio // and a video traf, and whichever comes first must see the same answer, or the other rolls // a fragment late and the two tracks stop agreeing on where segments fall. - if self.starts_segment(&moof, has_video, payload.len()) { + let has_video = self.tracks.values().any(|t| t.kind == Kind::Video); + if self.starts_segment(&moof, has_video, len.unwrap_or(usize::MAX)) { self.segment += 1; // A boundary the source declared is also an HLS segment boundary; one merely inferred // from a keyframe is not, since the default pacing already sees that group open. self.declared = self.pending_cut; self.pending_cut = false; } - - if self.moov.is_none() { - return Err(Error::NoMoov.into()); - } - - // Offsets include any boxes between moof and mdat, across decode calls too. - let data_start = self - .moof_span - .checked_sub(payload.len()) - .ok_or(Error::InvalidDataOffset)?; - let mut resolver = sample::Resolver::new(data_start, payload.len()); - - // Loop over all of the traf boxes in the moof. + let mut resolver = sample::Resolver::new(self.moof_span, len.unwrap_or(usize::MAX)); + resolver.streaming(); + let mut mdat = Mdat { + remaining: len, + position: 0, + tracks: Vec::new(), + chunks: std::collections::VecDeque::new(), + }; + // Validate every track before opening any payload writer. Otherwise a later traf's + // backwards run or reused bytes could fail after earlier tracks published data. + let mut anchored = false; for traf in &moof.traf { - let track_id = traf.tfhd.track_id; - if let Some(track) = self.text.get_mut(&track_id) { - track.extract( - super::text::Fragment { - traf, - data: &payload, - resolver: &mut resolver, - }, - &mut self.timebase, - )?; - self.initial_reservation = None; + let id = traf.tfhd.track_id; + if let Some(track) = self.text.get_mut(&id) { + for sample in track.prepare(traf, &mut resolver, &mut self.timebase)? { + mdat.chunks.push_back(Chunk { + range: sample.range.clone(), + output: ChunkOutput::Text { + id, + sample, + data: BytesMut::new(), + }, + }); + } + anchored = true; continue; } - let track = match self.tracks.get_mut(&track_id) { + let track = match self.tracks.get_mut(&id) { Some(track) => track, - // A fragment for a track `select` dropped: ignore it. - None if self.skipped.contains(&track_id) => continue, - None => return Err(Error::UnknownTrack(track_id).into()), + None if self.skipped.contains(&id) => continue, + None => return Err(Error::UnknownTrack(id).into()), }; - - // A fragment wholly before the edit or the first sync sample, such as priming samples or - // a mid-GOP join. Skipped before the segment is counted as opened, or the next keyframe - // would roll a second segment. + // A fragment wholly before the edit or the first sync sample publishes nothing. let Some(run) = track.run(traf, &mut resolver)? else { continue; }; - - // Shift onto the catalog clock. Anchor before releasing the reservation, so the first - // snapshot carries the final clock; the moov declared every track, so any track's - // fragment releases it. + // The moov declared every track, so any track's anchored fragment releases the catalog. let offset = self.timebase.anchor(run.earliest)?; - self.initial_reservation = None; - - let new_segment = track.segment != Some(self.segment); + anchored = true; let placement = Placement { offset, - new_segment, - // A group opening for a declared segment is an HLS segment boundary on this track. - declared: new_segment && self.declared, + new_segment: track.segment != Some(self.segment), + declared: track.segment != Some(self.segment) && self.declared, }; + let prefix = match track.output { + Output::Passthrough(_) => Some(fragment(traf, &moof, &run, &track.defaults)?), + _ => None, + }; + let index = mdat.tracks.len(); + for (sample, value) in run.samples.iter().enumerate() { + mdat.chunks.push_back(Chunk { + range: value.range.clone(), + output: ChunkOutput::Media { track: index, sample }, + }); + } + let size = prefix.as_ref().map_or(0, Bytes::len) + + run.samples.iter().map(|sample| sample.range.len()).sum::(); + mdat.tracks.push(PendingTrack { + size, + id, + run, + placement, + prefix, + writer: None, + opened: false, + }); + } + mdat.chunks + .make_contiguous() + .sort_unstable_by_key(|chunk| chunk.range.start); + if anchored { + self.initial_reservation = None; + } + Ok(mdat) + } - let opened = match &mut track.output { - Output::Passthrough(passthrough) => { - let fragment = fragment(traf, &moof, &run, &payload, &track.defaults)?; - passthrough.write(fragment, &run, track.kind, placement)? - } - output => unwrap(output, &run, &payload, track.timescale, placement)?, + fn write_mdat(&mut self, mdat: &mut Mdat, mut data: &[u8]) -> Result<()> { + while !data.is_empty() { + let Some(chunk) = mdat.chunks.front_mut() else { + mdat.position = mdat.position.checked_add(data.len()).ok_or(Error::InvalidDataOffset)?; + return Ok(()); }; - // Consuming the boundary on a dependent fragment would let its next sync sample - // advance the segment again before this track has opened the first one. - if opened { - track.segment = Some(self.segment); + if mdat.position < chunk.range.start { + let skip = (chunk.range.start - mdat.position).min(data.len()); + mdat.position += skip; + data = &data[skip..]; + continue; + } + let count = (chunk.range.end - mdat.position).min(data.len()); + let complete = mdat.position + count == chunk.range.end; + match &mut chunk.output { + ChunkOutput::Text { + id, + sample, + data: buffer, + } => { + buffer.extend_from_slice(&data[..count]); + if complete { + self.text + .get_mut(id) + .expect("planned text track") + .write(sample, buffer)?; + } + } + ChunkOutput::Media { + track: index, + sample: sample_index, + } => { + let pending = &mut mdat.tracks[*index]; + let track = self.tracks.get_mut(&pending.id).expect("planned media track"); + if pending.writer.is_none() { + if *sample_index == 0 { + pending.placement.new_segment = track.segment != Some(self.segment); + pending.placement.declared = pending.placement.new_segment && self.declared; + } + let (writer, opened) = match &mut track.output { + Output::Passthrough(output) => { + let prefix = pending.prefix.take().expect("one prefix per fragment"); + let (writer, opened) = + output.start(prefix, &pending.run, track.kind, pending.placement)?; + (Writer::Fragment(writer), opened) + } + output => { + let (sample, opened) = open_sample(output, pending, *sample_index, track.timescale)?; + (Writer::Sample(sample), opened) + } + }; + pending.writer = Some(writer); + if opened { + pending.opened = true; + track.segment = Some(self.segment); + } + } + pending.writer.as_mut().expect("opened writer").write(&data[..count])?; + if complete { + let last = *sample_index + 1 == pending.run.samples.len(); + match &mut track.output { + Output::Passthrough(output) if last => { + let Some(Writer::Fragment(writer)) = pending.writer.take() else { + unreachable!("passthrough tracks write whole fragments"); + }; + output.complete(writer, pending, track.kind)?; + } + Output::Passthrough(_) => {} + output => media!(output, producer => { + let Some(Writer::Sample(sample)) = pending.writer.take() else { + unreachable!("unwrapped tracks write one frame per sample"); + }; + producer.commit(sample)?; + if last { + producer.burst(pending.run.end.checked_sub(pending.run.earliest)?.into())?; + } + }, _ => unreachable!()), + } + } + } + } + mdat.position += count; + data = &data[count..]; + if complete { + mdat.chunks.pop_front(); + if mdat.chunks.is_empty() { + mdat.tracks.clear(); + } } } + Ok(()) + } +} + +// Framing never retains skipped payload; media bytes go straight into the transport's cache. +enum Body { + Skip(Option), + Metadata { + kind: mp4_atom::FourCC, + remaining: usize, + header_size: usize, + }, + Media(Mdat), +} + +struct Mdat { + remaining: Option, + position: usize, + tracks: Vec, + chunks: std::collections::VecDeque, +} + +impl Mdat { + fn abort(&mut self, err: moq_net::Error) { + for pending in &mut self.tracks { + match pending.writer.take() { + Some(Writer::Fragment(writer)) => { + let _ = writer.abort(err.clone()); + } + Some(Writer::Sample(sample)) => sample.abort(err.clone()), + None => {} + } + } + } +} + +struct PendingTrack { + size: usize, + id: u32, + run: Run, + placement: Placement, + prefix: Option, + writer: Option, + opened: bool, +} +/// The output frame a track's payload streams into. +enum Writer { + /// A passthrough track's whole per-track fragment. + Fragment(moq_net::frame::Producer), + /// One unwrapped sample. + Sample(Streamed), +} + +impl Writer { + fn write(&mut self, chunk: &[u8]) -> crate::Result<()> { + match self { + Self::Fragment(writer) => writer.write(chunk)?, + Self::Sample(sample) => sample.write(chunk)?, + } Ok(()) } } +struct Chunk { + range: std::ops::Range, + output: ChunkOutput, +} + +enum ChunkOutput { + Media { + track: usize, + sample: usize, + }, + Text { + id: u32, + sample: super::text::Sample, + data: BytesMut, + }, +} + /// Where a fragment lands on its track. #[derive(Clone, Copy)] struct Placement { @@ -1156,17 +1452,11 @@ impl Fmp4Track { } } -/// Re-encode `traf`'s samples as a standalone moof+mdat fragment addressing only its own bytes. +/// Re-encode the header of a standalone per-track fragment addressing only its own bytes. /// /// The mdat holds `run`'s samples back to back in decode order, whatever separated them in the /// source (padding, another track's runs), so each rewritten `data_offset` is the sizes before it. -fn fragment( - traf: &mp4_atom::Traf, - moof: &Moof, - run: &Run, - payload: &Bytes, - defaults: &sample::Defaults, -) -> Result { +fn fragment(traf: &mp4_atom::Traf, moof: &Moof, run: &Run, defaults: &sample::Defaults) -> Result { // Build a per-track moof containing only this traf, and a per-track mdat // with only the samples belonging to this track. let mut adjusted_moof = Moof { @@ -1205,13 +1495,20 @@ fn fragment( adjusted_moof.encode(&mut moof_buf)?; let new_moof_size = moof_buf.len(); + // A payload past 4 GiB takes the 16-byte largesize header. + let mut mdat = Vec::new(); + mp4_atom::Header { + kind: (*b"mdat").into(), + size: Some(run.samples.iter().map(|sample| sample.range.len()).sum()), + } + .encode(&mut mdat)?; + // Re-encode moof with corrected per-trun data_offset for the per-track fragment. // Each trun's data_offset points to the start of that run's data within the new mdat. - let mdat_header_size_new = 8u64; // 4 bytes size + 4 bytes 'mdat' - let mut cumulative_offset = 0u64; + let mut cumulative_offset = (new_moof_size + mdat.len()) as u64; for traf_mut in &mut adjusted_moof.traf { for trun_mut in &mut traf_mut.trun { - trun_mut.data_offset = Some((new_moof_size as u64 + mdat_header_size_new + cumulative_offset) as i32); + trun_mut.data_offset = Some(i32::try_from(cumulative_offset).map_err(|_| Error::InvalidDataOffset)?); // Advance past this trun's sample data let trun_data_size: u64 = trun_mut @@ -1225,24 +1522,9 @@ fn fragment( moof_buf.clear(); adjusted_moof.encode(&mut moof_buf)?; - - // Adjacent samples are copied as one span. - let mut data = Vec::with_capacity(run.samples.iter().map(|sample| sample.range.len()).sum()); - let mut span: Option> = None; - for sample in &run.samples { - match &mut span { - Some(span) if span.end == sample.range.start => span.end = sample.range.end, - _ => { - if let Some(span) = span.replace(sample.range.clone()) { - data.extend_from_slice(&payload[span]); - } - } - } - } - if let Some(span) = span { - data.extend_from_slice(&payload[span]); - } - Mdat { data }.encode(&mut moof_buf)?; + // Each data_offset was reserved above, so restating them keeps the moof's size. + debug_assert_eq!(moof_buf.len(), new_moof_size); + moof_buf.extend_from_slice(&mdat); Ok(Bytes::from(moof_buf)) } @@ -1283,13 +1565,18 @@ fn kept(traf: &mp4_atom::Traf, run: &Run) -> Result { } impl Passthrough { - /// Write one per-track `fragment` as a single frame, reporting whether it opened a group. - fn write(&mut self, fragment: Bytes, run: &Run, kind: Kind, placement: Placement) -> Result { + /// Open one per-track fragment's output frame, reporting whether it opened a group. + fn start( + &mut self, + prefix: Bytes, + run: &Run, + kind: Kind, + placement: Placement, + ) -> Result<(moq_net::frame::Producer, bool)> { // Carry the fragment's earliest presentation time as the frame timestamp, in the // track's native timescale and shifted onto the catalog clock. Consumers present the // fragment at it; `tfdt` only places the samples relative to each other. let timestamp = placement.offset.apply(run.earliest)?; - let end = placement.offset.apply(run.end)?; // Audio follows segment boundaries; video also waits for a sync first sample so a new // group's first fragment can be decoded independently. @@ -1303,7 +1590,7 @@ impl Passthrough { { return Err(crate::container::TimestampRewind { timestamp, floor }.into()); } - let mut group = if start_group { + let group = if start_group { if let Some(mut prev) = self.group.take() { prev.finish()?; } @@ -1315,8 +1602,6 @@ impl Passthrough { self.group.take().ok_or(Error::NoKeyframe)? }; - let sequence = group.sequence; - if start_group { // A group reopened mid-segment (after a `seek`) is no segment boundary. if placement.declared @@ -1329,11 +1614,27 @@ impl Passthrough { // descriptor didn't declare a bitrate). self.estimator.cut(Some(timestamp)); } - let fragment_len = fragment.len(); - group.write_frame(timestamp, fragment)?; + let size = prefix + .len() + .checked_add(run.samples.iter().map(|sample| sample.range.len()).sum()) + .ok_or(Error::InvalidDataOffset)?; + let mut writer = group.create_frame(moq_net::frame::Info { + timestamp: Some(timestamp), + size: size as u64, + })?; + writer.write(prefix)?; if start_group { self.start = Some(timestamp); } + Ok((writer, start_group)) + } + + fn complete(&mut self, writer: moq_net::frame::Producer, pending: &PendingTrack, kind: Kind) -> Result<()> { + let group = writer.finish()?; + let sequence = group.sequence; + let run = &pending.run; + let timestamp = pending.placement.offset.apply(run.earliest)?; + let end = pending.placement.offset.apply(run.end)?; // Audio is always independently decodable; video counts the IDRs it carries. let keyframes = match kind { @@ -1347,7 +1648,7 @@ impl Passthrough { self.group = Some(group); - self.estimator.write(timestamp, fragment_len); + self.estimator.write(timestamp, pending.size); let bytes = run.samples.iter().map(|sample| sample.range.len()).sum(); self.sent.add(run.samples.len() as u64, bytes, keyframes); if let Some(recorder) = self.recorder.as_mut() { @@ -1355,65 +1656,69 @@ impl Passthrough { } self.estimator.burst(end.checked_sub(timestamp)?.into()); self.publish_estimate()?; - Ok(start_group) + Ok(()) } } -/// Write each sample of `run` as its own frame on an unwrapped track. -fn unwrap( +/// Open one sample's frame without retaining its payload in the importer. +fn open_sample( output: &mut Output, - run: &Run, - payload: &Bytes, + pending: &PendingTrack, + index: usize, timescale: moq_net::Timescale, - placement: Placement, -) -> Result { - let earliest = placement.offset.apply(run.earliest)?; - let end = placement.offset.apply(run.end)?; +) -> Result<(Streamed, bool)> { let kind = if matches!(output, Output::Video(_)) { Kind::Video } else { Kind::Audio }; - + let run = &pending.run; + let placement = pending.placement; + let sample = &run.samples[index]; media!(output, producer => { - let mut opened = false; - for (index, sample) in run.samples.iter().enumerate() { - let keyframe = match kind { - Kind::Video => placement.keyframe(kind, sample), - Kind::Audio => (index == 0 && placement.keyframe(kind, sample)) || producer.needs_keyframe(), - }; - if keyframe { - if !opened && placement.declared { - producer.boundary(earliest); - } - opened = true; - } - if sample.pts > sample.dts { - producer.reorder(Timestamp::new(sample.pts - sample.dts, timescale)?); - } - let frame = crate::container::Frame { - timestamp: placement.offset.apply(sample.timestamp)?, - payload: payload.slice(sample.range.clone()), - keyframe, - duration: sample - .duration - .map(|duration| Timestamp::new(duration as u64, timescale)) - .transpose()?, - }; - producer.write(frame)?; + let keyframe = match kind { + Kind::Video => sample.sync, + Kind::Audio => (index == 0 && placement.new_segment) || producer.needs_keyframe(), + }; + if keyframe && placement.declared && !pending.opened { + producer.boundary(placement.offset.apply(run.earliest)?); } - - // The fragment arrived at once, so a consumer has to buffer its whole span. - producer.burst(end.checked_sub(earliest)?.into())?; - Ok(opened) + if sample.pts > sample.dts { + producer.reorder(Timestamp::new(sample.pts - sample.dts, timescale)?); + } + let timestamp = placement.offset.apply(sample.timestamp)?; + let duration = sample.duration.map(|duration| Timestamp::new(duration as u64, timescale)).transpose()?; + let streamed = producer.open(timestamp, keyframe, duration, sample.range.len())?; + Ok((streamed, keyframe)) }, _ => unreachable!("passthrough tracks write whole fragments")) } impl Import { /// Finish all tracks, flushing current groups. + /// + /// 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). 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(); + // A seek or discontinuity waiting on the truncated fragment has no fragment left to land on. + self.deferred = Deferred::default(); + let truncated = match self.body.take() { + Some(Body::Media(mut mdat)) if !mdat.chunks.is_empty() => { + mdat.abort(moq_net::Error::WrongSize); + true + } + _ => false, + }; + self.finish_tracks()?; + if truncated { + return Err(mp4_atom::Error::UnexpectedEof.into()); + } + Ok(()) + } + + fn finish_tracks(&mut self) -> Result<()> { for track in self.text.values_mut() { track.producer.finish()?; } @@ -1431,6 +1736,9 @@ impl Import { /// Unlike a drop, this leaves the tracks' HLS records to the catalog, whose abort fails them /// rather than ending them as a complete recording. pub fn abort(mut self, err: moq_net::Error) { + if let Some(Body::Media(mdat)) = &mut self.body { + mdat.abort(err.clone()); + } for track in std::mem::take(&mut self.text).into_values() { track.producer.abort(err.clone()); } @@ -1453,7 +1761,18 @@ impl Import { /// [`discontinuity`](Self::discontinuity). Text spends a group per cue, so a text track that /// is already past `sequence` continues from its own next group instead. Video resumes at its /// next sync sample, dropping the samples before it. + /// + /// 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 self.publishing() { + self.deferred.seek = Some(sequence); + return Ok(()); + } + self.apply_seek(sequence) + } + + fn apply_seek(&mut self, sequence: u64) -> Result<()> { for track in self.text.values_mut() { let next = track.producer.next_sequence(); track.producer.seek(sequence.max(next))?; @@ -1475,8 +1794,18 @@ impl Import { /// fragment's decode time is not compared with the last one on each track. A timeline that /// restarts below where it was still fails with /// [`TimestampRewind`](crate::container::TimestampRewind): a track refuses a group that - /// rewinds, before publishing any of it. + /// 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 self.publishing() { + self.deferred.discontinuity = true; + return Ok(()); + } + self.apply_discontinuity(); + Ok(()) + } + + fn apply_discontinuity(&mut self) { for track in self.text.values_mut() { track.end_decode_time = None; track.discontinuity = true; @@ -1484,14 +1813,40 @@ impl Import { for track in self.tracks.values_mut() { track.last_decode_time = None; } + } + + /// Whether a fragment has samples still to arrive, whose open frames hold their groups. + fn publishing(&self) -> bool { + matches!(&self.body, Some(Body::Media(mdat)) if !mdat.chunks.is_empty()) + } + + /// Apply the controls deferred while a fragment was publishing, in the order HLS issues them. + fn settle(&mut self) -> Result<()> { + let Deferred { seek, discontinuity } = std::mem::take(&mut self.deferred); + if discontinuity { + self.apply_discontinuity(); + } + if let Some(sequence) = seek { + self.apply_seek(sequence)?; + } Ok(()) } } +/// Controls called while a fragment was still publishing, applied once it has. +#[derive(Default)] +struct Deferred { + seek: Option, + discontinuity: bool, +} + impl Drop for Import { /// End each track, then retire its HLS records: every track takes a fresh name, so nothing /// publishes it again, as when an HLS import replaces this init segment's importer. fn drop(&mut self) { + if let Some(Body::Media(mdat)) = &mut self.body { + mdat.abort(moq_net::Error::Cancel); + } for track in self.text.values_mut() { if let Err(err) = track.producer.end() { tracing::warn!(%err, "failed to end a dropped fMP4 text track"); @@ -1509,3 +1864,392 @@ impl Drop for Import { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn importer() -> Import { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + Import::new(broadcast, catalog.reserve()) + } + + #[test] + fn skipped_boxes_do_not_retain_their_payload() { + for kind in [b"free", b"skip", b"sidx", b"emsg", b"uuid"] { + let mut import = importer(); + let mut header = Vec::new(); + mp4_atom::Header { + kind: (*kind).into(), + size: Some(1usize << 62), + } + .encode(&mut header) + .unwrap(); + for byte in &header { + import.decode(&[*byte]).unwrap(); + } + // Each call can be another segment: a never-ending ignored box must not + // pin that segment, or any earlier segment, in the importer's buffer. + for _ in 0..256 { + import.cut().unwrap(); + import.decode(&[0xA5; 1024]).unwrap(); + assert!(import.buffer.is_empty(), "{kind:?}: {} held bytes", import.buffer.len()); + } + } + } + + fn initialized(passthrough: bool) -> (Import, u32, moq_net::cache::Pool) { + use mp4_atom::Decode; + let mut info = moq_net::broadcast::Info::new(); + info.pool = moq_net::cache::Pool::new(moq_net::cache::Config::default().with_capacity(1024 * 1024)); + let pool = info.pool.clone(); + let mut broadcast = info.produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut import = Import::new(broadcast, catalog.reserve()).with_passthrough(passthrough); + let mut data = include_bytes!("test_data/bbb.mp4").as_slice(); + mp4_atom::Ftyp::decode(&mut data).unwrap(); + let mut moov = Moov::decode(&mut data).unwrap(); + moov.trak.retain(|trak| trak.mdia.hdlr.handler.as_ref() == b"soun"); + let id = moov.trak[0].tkhd.track_id; + import.init(moov).unwrap(); + (import, id, pool) + } + + // Only the headers are supplied. The declared mdat may never finish. + fn media_header(id: u32, decode_time: u64, offsets: &[usize], size: u32, len: Option) -> Vec { + let mut moof = Moof { + mfhd: Default::default(), + traf: vec![mp4_atom::Traf { + tfhd: mp4_atom::Tfhd { + track_id: id, + default_sample_size: Some(size), + default_sample_duration: Some(1024), + ..Default::default() + }, + tfdt: Some(mp4_atom::Tfdt { + base_media_decode_time: decode_time, + }), + trun: offsets + .iter() + .map(|_| mp4_atom::Trun { + data_offset: Some(0), + samples: vec![mp4_atom::TrunEntry::default()].into(), + }) + .collect(), + ..Default::default() + }], + }; + let mut header = Vec::new(); + mp4_atom::Header { + kind: (*b"mdat").into(), + size: len, + } + .encode(&mut header) + .unwrap(); + let mut data = Vec::new(); + moof.encode(&mut data).unwrap(); + let start = data.len() + header.len(); + for (trun, offset) in moof.traf[0].trun.iter_mut().zip(offsets) { + trun.data_offset = Some((start + offset) as i32); + } + data.clear(); + moof.encode(&mut data).unwrap(); + data.extend_from_slice(&header); + data + } + + #[test] + fn backwards_ranges_fail_before_publishing() { + use futures::FutureExt; + for passthrough in [false, true] { + let (mut import, id, _) = initialized(passthrough); + let mut consumer = match &import.tracks[&id].output { + Output::Audio(output) => output.consume(), + Output::Passthrough(output) => output.track.subscribe(None), + _ => unreachable!(), + }; + let header = media_header(id, 0, &[4, 0], 4, Some(8)); + 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(); + import.finish().unwrap(); + assert!(consumer.recv_group().now_or_never().is_some()); + } + } + + fn box_header(kind: &[u8; 4], size: Option) -> Vec { + let mut header = Vec::new(); + mp4_atom::Header { + kind: (*kind).into(), + size, + } + .encode(&mut header) + .unwrap(); + header + } + + #[test] + fn oversized_metadata_is_refused_at_its_header() { + for kind in [b"moov", b"moof"] { + for size in [super::super::MAX_HELD_BYTES + 1, 1usize << 62] { + let mut import = importer(); + let header = box_header(kind, Some(size)); + let (last, prefix) = header.split_last().unwrap(); + for byte in prefix { + import.decode(&[*byte]).unwrap(); + } + let err = import.decode(&[*last]).unwrap_err(); + assert!( + 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()); + } + } + } + } + + #[test] + fn metadata_at_the_bound_holds_only_what_arrived() { + let mut import = importer(); + import + .decode(&box_header(b"moof", Some(super::super::MAX_HELD_BYTES))) + .unwrap(); + import.decode(&[0xA5; 1024]).unwrap(); + assert!(matches!(import.body, Some(Body::Metadata { .. }))); + assert_eq!(import.buffer.len(), 1024); + } + + // A moof whose runs carry no per-sample fields, each declaring `count` samples in a few bytes. + fn inherited_moof(id: u32, counts: &[u32]) -> Vec { + let moof = Moof { + mfhd: Default::default(), + traf: vec![mp4_atom::Traf { + tfhd: mp4_atom::Tfhd { + track_id: id, + default_sample_size: Some(1), + default_sample_duration: Some(1024), + ..Default::default() + }, + tfdt: Some(mp4_atom::Tfdt { + base_media_decode_time: 0, + }), + trun: counts + .iter() + .map(|&count| mp4_atom::Trun { + data_offset: None, + samples: mp4_atom::TrunSamples::Inherited { + count, + first_flags: None, + }, + }) + .collect(), + ..Default::default() + }], + }; + let mut data = Vec::new(); + moof.encode(&mut data).unwrap(); + data + } + + #[test] + fn huge_trun_sample_counts_are_refused() { + let max = super::super::MAX_FRAGMENT_SAMPLES; + let half = (max / 2) as u32; + for counts in [vec![u32::MAX], vec![half, half + 1], vec![u32::MAX; 4]] { + let (mut import, id, _) = initialized(false); + let moof = inherited_moof(id, &counts); + assert!(moof.len() < 256, "the declaration costs only a few bytes"); + let err = import.decode(&moof).unwrap_err(); + let total: u64 = counts.iter().map(|&count| count as u64).sum(); + assert!( + matches!(err, crate::Error::Cmaf(Error::TooManySamples(count)) if count == total), + "{counts:?}: {err:?}" + ); + 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(); + import.finish().unwrap(); + } + + // At the bound, the moof is accepted. + let (mut import, id, _) = initialized(false); + import.decode(&inherited_moof(id, &[half, half])).unwrap(); + assert!(import.moof.is_some()); + } + + #[test] + fn media_is_owned_by_the_output_before_mdat_finishes() { + const SIZE: usize = 64 * 1024; + for passthrough in [false, true] { + for len in [Some(1usize << 62), None] { + let (mut import, id, pool) = initialized(passthrough); + let header = media_header(id, 0, &[0, SIZE + 17], SIZE as u32, len); + for byte in &header { + import.decode(&[*byte]).unwrap(); + } + import.decode(&[0xA5; 1024]).unwrap(); + assert!(import.buffer.is_empty()); + let Some(Body::Media(mdat)) = &import.body else { + panic!("missing mdat"); + }; + assert!( + mdat.tracks[0].writer.is_some(), + "sample must reach its output incrementally" + ); + assert!( + pool.used() >= 1024, + "streamed bytes must be charged to the output cache" + ); + let remaining = SIZE * 2 + 17 - 1024; + for chunk in vec![0xA5; remaining].chunks(509) { + import.decode(chunk).unwrap(); + } + let held = pool.used(); + for _ in 0..256 { + import.cut().unwrap(); + import.decode(&[0xCC; 1024]).unwrap(); + assert!(import.buffer.is_empty()); + assert_eq!( + pool.used(), + held, + "unreferenced payload must not enter the output cache" + ); + } + import.finish().unwrap(); + } + } + } + + fn replay(output: &Output<()>) -> moq_net::track::Subscriber { + let replay = moq_net::track::Subscription::default().with_max_delay(std::time::Duration::from_secs(30)); + match output { + Output::Audio(output) => output.subscribe(replay), + Output::Passthrough(output) => output.track.subscribe(replay), + _ => unreachable!(), + } + } + + /// Each closed group's sequence and frame count, or `None` for a group that failed. + fn groups(consumer: &mut moq_net::track::Subscriber) -> Vec<(u64, Option)> { + use futures::FutureExt; + let mut groups = Vec::new(); + while let Some(mut group) = consumer.recv_group().now_or_never().and_then(|r| r.ok().flatten()) { + let mut frames = Some(0); + loop { + match group.next_frame().now_or_never() { + Some(Ok(Some(_))) => frames = frames.map(|count| count + 1), + Some(Ok(None)) => break, + Some(Err(_)) => { + frames = None; + break; + } + None => panic!("group {} left open", group.sequence), + } + } + groups.push((group.sequence, frames)); + } + groups + } + + #[test] + fn seek_waits_for_the_publishing_fragment() { + const SIZE: usize = 64; + for passthrough in [false, true] { + let (mut import, id, _) = initialized(passthrough); + let mut consumer = replay(&import.tracks[&id].output); + let sample = [0xA5; SIZE]; + import + .decode(&media_header(id, 0, &[0], SIZE as u32, Some(SIZE))) + .unwrap(); + import.decode(&sample[..SIZE / 2]).unwrap(); + // The open frame holds group 0, so neither control may close it underneath the sample. + import.discontinuity().unwrap(); + import.seek(100).unwrap(); + import.decode(&sample[SIZE / 2..]).unwrap(); + assert_eq!(groups(&mut consumer), [(0, Some(1))], "passthrough: {passthrough}"); + // The discontinuity lets the next fragment restate the decode time. + import + .decode(&media_header(id, 0, &[0], SIZE as u32, Some(SIZE))) + .unwrap(); + import.decode(&sample).unwrap(); + import.finish().unwrap(); + assert_eq!(groups(&mut consumer), [(100, Some(1))], "passthrough: {passthrough}"); + } + } + + #[test] + fn truncated_sample_fails_its_group_and_finishes_the_track() { + const SIZE: usize = 64; + for passthrough in [false, true] { + let (mut import, id, _) = initialized(passthrough); + let mut consumer = replay(&import.tracks[&id].output); + import + .decode(&media_header(id, 0, &[0], SIZE as u32, Some(SIZE))) + .unwrap(); + import.decode(&[0xA5; SIZE]).unwrap(); + import + .decode(&media_header(id, 1024, &[0], SIZE as u32, Some(SIZE))) + .unwrap(); + import.decode(&[0xA5; SIZE / 2]).unwrap(); + let err = import.finish().unwrap_err(); + assert!(matches!(err, crate::Error::Mp4(_)), "{err:?}"); + assert_eq!( + groups(&mut consumer), + [(0, Some(1)), (1, None)], + "passthrough: {passthrough}" + ); + let finished = match &import.tracks[&id].output { + Output::Audio(output) => output.finished(), + Output::Passthrough(output) => output.track.finished(), + _ => unreachable!(), + }; + assert!(finished.is_some(), "passthrough: {passthrough}"); + } + } + + #[test] + fn refused_fragment_discards_its_mdat() { + const SIZE: usize = 64; + for passthrough in [false, true] { + let (mut import, id, _) = initialized(passthrough); + let mut consumer = replay(&import.tracks[&id].output); + import + .decode(&media_header(id, 48_000, &[0], SIZE as u32, Some(SIZE))) + .unwrap(); + import.decode(&[0xA5; SIZE]).unwrap(); + // A new timeline that restarts below the published one is refused at its first sample. + import.discontinuity().unwrap(); + 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(); + assert_eq!( + groups(&mut consumer), + [(0, Some(1)), (1, 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 319dde90631..cf4192072c9 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1078,9 +1078,8 @@ async fn unwrapped_samples_present_at_their_composition_time() { ); } -/// bbb's init with `entries` as the edit list of track `track_id`. -/// Once the video group is open, the keyframe scan for a segment boundary stops at the mdat -/// rather than walking a forged moof's field-less runs of `u32::MAX` non-sync samples. +/// Once the video group is open, a forged moof's field-less runs of `u32::MAX` non-sync samples +/// are refused at the moof, before the keyframe scan or sample planning walks them. #[test] fn fieldless_runs_do_not_stall_the_segment_scan() { let (_, (video_id, video_scale), _) = bbb_init(); @@ -1131,7 +1130,7 @@ fn fieldless_runs_do_not_stall_the_segment_scan() { assert!( matches!( fmp4.decode(&forged), - Err(crate::Error::Cmaf(super::Error::SampleRangeOutOfBounds { .. })) + Err(crate::Error::Cmaf(super::Error::TooManySamples(_))) ), "{trafs} trafs, passthrough {passthrough}" ); @@ -3221,6 +3220,64 @@ 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. +#[test] +fn oversized_text_sample_is_refused_before_staging() { + let size = super::MAX_HELD_BYTES + 1; + let moof = mp4_atom::Moof { + mfhd: mp4_atom::Mfhd { sequence_number: 1 }, + traf: vec![mp4_atom::Traf { + tfhd: mp4_atom::Tfhd { + track_id: 1, + default_sample_duration: Some(1000), + ..Default::default() + }, + tfdt: Some(mp4_atom::Tfdt { + base_media_decode_time: 0, + }), + trun: vec![mp4_atom::Trun { + data_offset: None, + samples: vec![mp4_atom::TrunEntry { + size: Some(size as u32), + ..Default::default() + }] + .into(), + }], + ..Default::default() + }], + }; + let mut header = Vec::new(); + moof.encode(&mut header).unwrap(); + mp4_atom::Header { + kind: b"mdat".into(), + size: Some(size), + } + .encode(&mut header) + .unwrap(); + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut importer = super::Import::new(broadcast, catalog.reserve()); + importer.decode(&webvtt_init("WEBVTT")).unwrap(); + let err = importer.decode(&header).unwrap_err(); + assert!( + 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; + } + let cue = subtitle_box(b"vttc", &subtitle_box(b"payl", b"hello")); + importer.decode(&webvtt_fragment(0, &[cue], 1000)).unwrap(); + importer.finish().unwrap(); + assert_eq!(catalog.snapshot().text.renditions.len(), 1); +} + /// The WebVTT `Kind` header sets the role, and lone CR line terminators are accepted. #[tokio::test(start_paused = true)] async fn webvtt_kind_and_lone_cr_terminators() { diff --git a/rs/moq-mux/src/container/fmp4/mod.rs b/rs/moq-mux/src/container/fmp4/mod.rs index b6cd27dab9d..b14ff7d172f 100644 --- a/rs/moq-mux/src/container/fmp4/mod.rs +++ b/rs/moq-mux/src/container/fmp4/mod.rs @@ -24,6 +24,22 @@ pub use muxer::*; /// A major version bump of `mp4_atom` is a breaking change for moq-mux. pub use mp4_atom; +/// The most input [`Import`] holds whole: a `moov` or `moof` body, or one text sample. +/// +/// Metadata parses and text rewrites only once their last byte arrives, so a larger declaration +/// is refused at its header rather than retaining later input (across HLS segments, too) until +/// it completes. It matches the transport's per-group bound, +/// [`MAX_CACHE_BYTES`](moq_net::group::MAX_CACHE_BYTES). Real fMP4 metadata is kilobytes. Media +/// samples stream into their frames and have no such bound. +pub const MAX_HELD_BYTES: usize = moq_net::group::MAX_CACHE_BYTES as usize; + +/// The most samples [`Import`] plans for one `moof`. +/// +/// Every sample's byte range is resolved before any of the fragment publishes. A `moof` at +/// [`MAX_HELD_BYTES`] holds this many 16-byte `trun` entries; a run without per-sample fields +/// could otherwise declare billions of samples in a few bytes. +pub const MAX_FRAGMENT_SAMPLES: u64 = (MAX_HELD_BYTES / 16) as u64; + #[cfg(test)] mod export_test; #[cfg(test)] @@ -191,6 +207,31 @@ pub enum Error { #[error("invalid data offset")] InvalidDataOffset, + /// Streaming import requires each track's sample bytes in decode order. + #[error("backwards per-track data offset is unsupported for streaming import")] + BackwardsDataOffset, + + /// A `moov` or `moof` must be held whole before it parses, so one declaring more than + /// [`MAX_HELD_BYTES`] is refused at its header, before any of its body is buffered. + #[error("{kind} declares {size} bytes, more than the {MAX_HELD_BYTES} an importer holds")] + MetadataTooLarge { + /// The box type. + kind: mp4_atom::FourCC, + /// The body size its header declares. + size: usize, + }, + + /// A text sample is rewritten whole, so one larger than [`MAX_HELD_BYTES`] is refused + /// before any of its bytes are staged. + #[error("text sample of {0} bytes is more than the {MAX_HELD_BYTES} an importer holds")] + TextSampleTooLarge(usize), + + /// A fragment's samples are planned before any of them publish, so a `moof` declaring more + /// than [`MAX_FRAGMENT_SAMPLES`] is refused. A `trun` without per-sample fields declares its + /// count in four bytes, so the byte bound alone does not limit it. + #[error("moof declares {0} samples, more than the {MAX_FRAGMENT_SAMPLES} an importer plans")] + TooManySamples(u64), + /// Absolute file offsets cannot be resolved in a standalone fragment. #[error("absolute base_data_offset is unsupported; use moof-relative offsets")] UnsupportedBaseDataOffset, diff --git a/rs/moq-mux/src/container/fmp4/sample.rs b/rs/moq-mux/src/container/fmp4/sample.rs index 80588c1ccf4..bceb192a73a 100644 --- a/rs/moq-mux/src/container/fmp4/sample.rs +++ b/rs/moq-mux/src/container/fmp4/sample.rs @@ -79,6 +79,7 @@ pub(crate) struct Resolver { data_start: usize, len: usize, claimed: Claimed, + forward: Option>, } /// A sample whose byte range and duration have been validated. @@ -94,9 +95,15 @@ impl Resolver { data_start, len, claimed: Claimed::default(), + forward: None, } } + // A sequential track writer cannot retain an earlier byte range for a later frame. + pub fn streaming(&mut self) { + self.forward = Some(Default::default()); + } + /// Walk samples without allocating from the untrusted sample count. pub fn resolve<'a>( &'a mut self, @@ -147,6 +154,13 @@ impl Resolver { } let range = offset..end; self.claimed.take(range.clone())?; + if let Some(forward) = &mut self.forward { + let previous = forward.entry(traf.tfhd.track_id).or_default(); + if range.start < *previous { + return Err(super::Error::BackwardsDataOffset); + } + *previous = range.end; + } if sample.duration.is_none() && remaining > 1 { return Err(super::Error::MissingSampleDuration); } diff --git a/rs/moq-mux/src/container/fmp4/text.rs b/rs/moq-mux/src/container/fmp4/text.rs index 45858382a6f..dad774b6792 100644 --- a/rs/moq-mux/src/container/fmp4/text.rs +++ b/rs/moq-mux/src/container/fmp4/text.rs @@ -44,67 +44,78 @@ pub(super) struct Track { pub discontinuity: bool, } -pub(super) struct Fragment<'a> { - pub traf: &'a mp4_atom::Traf, - pub data: &'a [u8], - /// Resolve sample ranges shared with the moof's other tracks. - pub resolver: &'a mut sample::Resolver, +pub(super) struct Sample { + pub range: std::ops::Range, + source: Timestamp, + timestamp: Timestamp, + duration: Timestamp, } impl Track { - pub fn extract( + pub fn prepare( &mut self, - input: Fragment<'_>, + traf: &mp4_atom::Traf, + resolver: &mut sample::Resolver, timebase: &mut crate::catalog::Timebase, - ) -> crate::Result<()> { - let traf = input.traf; + ) -> crate::Result> { let mut dts = traf.tfdt.as_ref().ok_or(Error::MissingTfdt)?.base_media_decode_time; if self.end_decode_time.is_some_and(|end| dts < end) { return Err(Error::InvalidSubtitle("subtitle fragment overlaps the previous one").into()); } - if traf.trun.is_empty() { - return Err(Error::MissingTrun.into()); - } - if self.discontinuity { - self.producer.discontinuity()?; - self.discontinuity = false; - } - for resolved in input.resolver.resolve(traf, self.defaults)? { + let mut samples = Vec::new(); + for resolved in resolver.resolve(traf, self.defaults)? { let sample::Resolved { sample, cts, range } = resolved?; + // A text sample is staged whole before it is rewritten. + if range.len() > super::MAX_HELD_BYTES { + return Err(Error::TextSampleTooLarge(range.len()).into()); + } let duration = sample.duration.ok_or(Error::MissingSampleDuration)?; let pts = dts.checked_add_signed(cts).ok_or(Error::PtsOverflow)?; let duration = Timestamp::new(duration as u64, self.timescale)?; - let data = &input.data[range]; dts = dts.checked_add(duration.value()).ok_or(Error::PtsOverflow)?; - - // A cue starting before the edit is dropped, like a media sample. let Some(presented) = self.edit.present(pts)? else { continue; }; - // TTML maps the document's media times from `source`, so they move with the edit too. - let source = Timestamp::new(pts, self.timescale)?; - let timestamp = timebase.shift(Timestamp::new(presented, self.timescale)?)?; - let end = timestamp.checked_add(duration)?; - let payload = match &self.format { - Format::Vtt { header } => vtt(header, data, timestamp, end)?, - // The frame timestamp is serialized in hang's scale, which the body's begin must match. - Format::Ttml => ttml::rewrite( - data, - source, - timestamp.convert(hang::container::TIMESCALE)?..end.convert(hang::container::TIMESCALE)?, - ) - .map_err(Error::InvalidSubtitle)? - .into(), - }; - self.producer.write(crate::container::Frame { - timestamp, - duration: Some(duration), - keyframe: true, - payload, - })?; - self.producer.cut(None)?; + samples.push(Sample { + range, + source: Timestamp::new(pts, self.timescale)?, + timestamp: timebase.shift(Timestamp::new(presented, self.timescale)?)?, + duration, + }); } self.end_decode_time = Some(dts); + Ok(samples) + } + + pub fn write(&mut self, sample: &Sample, data: &[u8]) -> crate::Result<()> { + if self.discontinuity { + self.producer.discontinuity()?; + self.discontinuity = false; + } + let Sample { + source, + timestamp, + duration, + .. + } = *sample; + let end = timestamp.checked_add(duration)?; + let payload = match &self.format { + Format::Vtt { header } => vtt(header, data, timestamp, end)?, + Format::Ttml => ttml::rewrite( + data, + source, + timestamp.convert(hang::container::TIMESCALE)?..end.convert(hang::container::TIMESCALE)?, + ) + .map_err(Error::InvalidSubtitle)? + .into(), + }; + self.producer.write(crate::container::Frame { + timestamp, + duration: Some(duration), + keyframe: true, + payload, + })?; + self.producer.cut(None)?; Ok(()) } }