From 63ffa35db7c547620ea6971e7820474136aab60b Mon Sep 17 00:00:00 2001 From: Frando Date: Wed, 2 Sep 2026 02:14:19 +0200 Subject: [PATCH 1/6] fix(moq-mux): roll fMP4 audio fragments at the video GOP boundary `fmp4::Export` rolled a fragment on a video keyframe or on the optional `fragment_duration` cap. An audio track has neither trigger with the default settings, so on an A/V broadcast its samples accumulated in the track buffer until the track finished: `moq export fmp4` produced a file whose every `traf` carried the video track id and whose audio arrived, if at all, as one trailing fragment. The buffer also grew for the length of the recording. Roll every audio track when a video keyframe rolls a video fragment, so the fragments of a broadcast cover the same spans and a player seeking to one finds both tracks. That follows what `mkv::Export` already does with a single cluster shared across tracks, rather than giving audio a default time cap, which would produce fragments on a schedule unrelated to the video and seek badly. A GOP boundary now yields more than one fragment, so they queue in `outgoing` and come back in ascending first-sample order, video first on a tie. Each video track still rolls only on its own keyframes, so a simulcast ladder whose keyframes do not coincide keeps every video fragment independently decodable. A broadcast with no video keeps the existing per-frame audio fragmenting, and audio that outlives a finished video track falls back to it instead of buffering again. (cherry picked from commit c0f3a4f528e8ea67240059d0eb30a00a43d1c40f) --- rs/moq-mux/src/container/fmp4/export.rs | 81 +++++++- rs/moq-mux/src/container/fmp4/export_test.rs | 187 ++++++++++++++++++- rs/moq-mux/src/container/test_util.rs | 20 +- 3 files changed, 275 insertions(+), 13 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index fb182ae8f5..1b466437d4 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::task::Poll; use std::time::Duration; @@ -24,8 +24,10 @@ use moq_net::Timestamp; /// Use [`next`](Self::next) to pull byte chunks: the first call returns the merged /// init segment (ftyp + multi-track moov), subsequent calls return moof+mdat /// fragments. By default each video fragment covers one GOP (rolled over on -/// keyframes); [`with_fragment_duration`](Self::with_fragment_duration) caps the -/// fragment duration for downstream consumers that throttle by fragment rate. +/// keyframes) and every audio track rolls at the same boundaries, so the +/// fragments of a broadcast line up in time; +/// [`with_fragment_duration`](Self::with_fragment_duration) caps the fragment +/// duration for downstream consumers that throttle by fragment rate. /// Returns `None` when the broadcast ends. /// /// [`next_fragment`](Self::next_fragment) returns the same bytes wrapped in a @@ -49,6 +51,11 @@ pub struct Export { /// Set after the init segment has been emitted; subsequent catalog updates only /// (un)subscribe tracks without re-emitting init. init_emitted: bool, + + /// Fragments encoded but not yet returned. A video GOP boundary rolls every + /// audio track alongside the video one, which produces more than one fragment + /// at a time; they're returned in ascending first-sample order. + outgoing: VecDeque, } /// One emitted CMAF chunk: either the init segment or a moof+mdat fragment, @@ -77,14 +84,16 @@ struct Fmp4Track { pending: Option, /// Frames accumulated for the current fragment. Flushed as a single - /// moof+mdat on the next keyframe (video) or duration cap. + /// moof+mdat at a GOP boundary (a keyframe on this track for video, the + /// keyframe of any video track for audio) or on the duration cap. buffer: Vec, /// Whether the first frame of the current `buffer` was a keyframe, i.e. the /// fragment it produces can start an HLS segment. Meaningless for audio. buffer_independent: bool, - /// True if this track is video. Video tracks roll fragments on keyframes. + /// True if this track is video. Video tracks roll fragments on their own + /// keyframes and drag the audio tracks along with them. is_video: bool, /// True for Opus audio, whose packets carry their duration in the TOC byte. @@ -118,6 +127,7 @@ impl Export { tracks: HashMap::new(), catalog_snapshot: None, init_emitted: false, + outgoing: VecDeque::new(), } } @@ -132,8 +142,9 @@ impl Export { /// Cap the fragment (moof+mdat) duration. /// - /// By default video fragments roll over on each keyframe (one fragment - /// per GOP); audio-only tracks emit one fragment per sample. Setting this + /// By default video fragments roll over on each keyframe (one fragment per + /// GOP) and audio rolls with them; a broadcast with no video at all emits + /// one audio fragment per sample, having no GOP to follow. Setting this /// caps each fragment to roughly `duration` of frames, useful for /// downstream consumers that throttle by fragment rate. [`Duration::ZERO`] /// emits one fragment per frame (the historical behavior); otherwise the @@ -174,6 +185,11 @@ impl Export { // track buffers a whole GOP, and stack use must stay flat however many // samples that is. loop { + // 0. Return anything a previous GOP boundary rolled but couldn't hand back yet. + if let Some(fragment) = self.outgoing.pop_front() { + return Poll::Ready(Ok(Some(fragment))); + } + // 1. Drain catalog updates and (un)subscribe tracks accordingly. while let Some(catalog) = self.catalog.as_mut() { match catalog.poll_next(waiter)? { @@ -263,7 +279,11 @@ impl Export { // no keyframe will ever roll the fragment. These never depend on the // successor, so emit immediately instead of buffering the frame until // the next one flushes it. - let has_video = self.tracks.values().any(|t| t.is_video); + // + // A finished video track can't roll anything either, so audio that + // outlives the video falls back to per-frame rather than buffering + // until it finishes too. + let has_video = self.tracks.values().any(|t| t.is_video && !t.finished); let per_frame = frag == Some(Duration::ZERO) || (frag.is_none() && !has_video); let track = self.tracks.get_mut(&name).unwrap(); let frame = track.pending.take().unwrap(); @@ -281,12 +301,27 @@ impl Export { return Poll::Ready(Ok(Some(fragment))); } if should_flush(track, &frame, frag) { + // A video keyframe closes a GOP for the whole presentation, not + // just for this track: audio has no keyframes of its own, so it + // rolls here to keep every track's fragments covering the same + // span. Without it an audio track would buffer until it finished. + let gop = track.is_video && frame.keyframe; + let start = track.buffer[0].timestamp; let frames = std::mem::take(&mut track.buffer); let fragment = emit_fragment(track, frames, Some(&frame))?; // The flushed run is done; the incoming frame opens the next buffer. track.buffer_independent = frame.keyframe; track.buffer.push(frame); - return Poll::Ready(Ok(Some(fragment))); + if !gop { + return Poll::Ready(Ok(Some(fragment))); + } + // Video first, so a tie at the same start puts the track that + // drove the boundary ahead of the audio that followed it. + let mut rolled = vec![(start, fragment)]; + rolled.extend(self.flush_audio()?); + rolled.sort_by_key(|(start, _)| *start); + self.outgoing.extend(rolled.into_iter().map(|(_, fragment)| fragment)); + return Poll::Ready(Ok(self.outgoing.pop_front())); } if track.buffer.is_empty() { track.buffer_independent = frame.keyframe; @@ -331,6 +366,30 @@ impl Export { } } + /// Encode every audio track's buffered run as a fragment, tagged with the + /// timestamp of its first sample so the caller can order it against the + /// video fragment it accompanies. + /// + /// Called only from a video GOP boundary, so no audio track is the one + /// currently being appended to. + fn flush_audio(&mut self) -> Result> { + let mut fragments = Vec::new(); + for track in self.tracks.values_mut() { + if track.is_video || track.buffer.is_empty() { + continue; + } + let start = track.buffer[0].timestamp; + let frames = std::mem::take(&mut track.buffer); + // The pending frame is this track's next sample, so it bounds the + // duration of the run's last one. + let successor = track.pending.take(); + let fragment = emit_fragment(track, frames, successor.as_ref())?; + track.pending = successor; + fragments.push((start, fragment)); + } + Ok(fragments) + } + fn update_catalog(&mut self, catalog: &Catalog) -> Result<()> { // A rendition we can't parse is ignored rather than failing the whole export. let mut catalog = catalog.clone(); @@ -556,7 +615,9 @@ pub(crate) fn extract_init( /// Should we flush `track.buffer` before appending the incoming `frame`? /// Triggers on a video keyframe (one fragment per GOP) or the duration cap. -/// Per-frame modes never buffer and are handled before this check. +/// Audio has no keyframe of its own and is rolled by the video track instead, +/// via [`Export::flush_audio`]. Per-frame modes never buffer and are handled +/// before this check. fn should_flush(track: &Fmp4Track, frame: &Frame, fragment_duration: Option) -> bool { if track.buffer.is_empty() { return false; diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index 8f77c55779..d94f150bb4 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -2,7 +2,7 @@ use std::io::Cursor; -use bytes::BytesMut; +use bytes::{Bytes, BytesMut}; use mp4_atom::{DecodeMaybe, Encode}; use crate::container::test_util::{Live, PPS, SPS, raw_frame, video_frame}; @@ -771,6 +771,191 @@ fn a_long_gop_does_not_cost_a_stack_frame_per_sample() { run.join().expect("a long GOP overflowed the stack"); } +/// An A/V broadcast must roll its audio fragments alongside the video ones. +/// +/// Audio has no keyframes, so with the default (per-GOP) fragmenting it has no +/// rollover trigger of its own: its samples used to pile up in the track buffer +/// and land in a single `traf` after the video track had already finished, which +/// left a player seeking to a video fragment with no audio to go with it. +#[tokio::test(start_paused = true)] +async fn audio_fragments_roll_with_the_video_gop() { + use hang::catalog::{AudioCodec, AudioConfig, Container}; + + let mut live = Live::avc3(); + let mut audio = live.add_track(".opus", |catalog, name| { + let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); + config.container = Container::Legacy; + catalog.lock().audio.renditions.insert(name, config); + }); + + // Three GOPs of three 30 fps frames, keyframes at 0, 99 ms and 198 ms. + for i in 0..9u64 { + live.track.write(video_frame(i * 33_000, i % 3 == 0)).unwrap(); + } + // 20 ms Opus packets (TOC 0x08) covering the same span and a little past it. + for i in 0..12u64 { + audio + .write(raw_frame(i * 20_000, &[0x08, 0xaa, 0xbb, 0xcc], true)) + .unwrap(); + } + live.track.finish().unwrap(); + audio.finish().unwrap(); + + let mut exporter = crate::container::fmp4::Export::new(live.source(), live.catalog_stream().await); + let init = fragment_now(&mut exporter).await; + assert!(init.init); + let (video_id, audio_id) = track_ids(&init.data); + + // Every fragment, tagged with the track it belongs to and its sample count. + let fragments: Vec<(u32, usize)> = drain_now(&mut exporter) + .await + .iter() + .map(|fragment| { + let trafs = traf_samples(&fragment.data); + assert_eq!(trafs.len(), 1, "expected one traf per fragment"); + trafs[0] + }) + .collect(); + + // Audio rolls with the video GOP, so the two tracks alternate: five 20 ms + // packets beside each three-frame GOP, then the tail of both tracks. + assert_eq!( + fragments, + vec![ + (video_id, 3), + (audio_id, 5), + (video_id, 3), + (audio_id, 5), + (video_id, 3), + (audio_id, 2), + ], + ); +} + +/// A simulcast ladder rolls each video track on its own keyframes: one +/// rendition's GOP boundary must not split another's, or the second rendition's +/// fragments stop being independently decodable. Audio rolls at whichever +/// boundary comes first, so it still never accumulates past one GOP. +#[tokio::test(start_paused = true)] +async fn simulcast_keyframes_only_roll_their_own_video_track() { + use hang::catalog::{AudioCodec, AudioConfig, Container, H264, VideoConfig}; + + let mut live = Live::avc3(); + let mut fast = live.add_track(".avc3", |catalog, name| { + let mut config = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: true, + }); + config.coded_width = Some(320); + config.coded_height = Some(240); + config.framerate = Some(30.0); + config.container = Container::Legacy; + catalog.lock().video.renditions.insert(name, config); + }); + let mut audio = live.add_track(".opus", |catalog, name| { + let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); + config.container = Container::Legacy; + catalog.lock().audio.renditions.insert(name, config); + }); + + // The same 30 fps cadence on both video tracks, but a three-frame GOP on one + // and a two-frame GOP on the other, so their keyframes rarely coincide. + for i in 0..9u64 { + live.track.write(video_frame(i * 33_000, i % 3 == 0)).unwrap(); + } + for i in 0..8u64 { + fast.write(video_frame(i * 33_000, i % 2 == 0)).unwrap(); + } + for i in 0..12u64 { + audio + .write(raw_frame(i * 20_000, &[0x08, 0xaa, 0xbb, 0xcc], true)) + .unwrap(); + } + live.track.finish().unwrap(); + fast.finish().unwrap(); + audio.finish().unwrap(); + + let mut exporter = crate::container::fmp4::Export::new(live.source(), live.catalog_stream().await); + let init = fragment_now(&mut exporter).await; + assert!(init.init); + let audio_id = track_ids(&init.data).1; + + let mut per_track: std::collections::BTreeMap> = std::collections::BTreeMap::new(); + for fragment in drain_now(&mut exporter).await { + let trafs = traf_samples(&fragment.data); + assert_eq!(trafs.len(), 1, "expected one traf per fragment"); + let (track_id, samples) = trafs[0]; + assert!( + fragment.independent, + "track {track_id} fragment of {samples} samples was split off a keyframe", + ); + per_track.entry(track_id).or_default().push(samples); + } + + // Each video track keeps its own GOP length; the audio track rolls at every + // boundary either of them draws, plus the tail both leave behind. + let mut video: Vec> = per_track + .iter() + .filter(|(id, _)| **id != audio_id) + .map(|(_, samples)| samples.clone()) + .collect(); + video.sort(); + assert_eq!(video, vec![vec![2, 2, 2, 2], vec![3, 3, 3]]); + assert_eq!(per_track[&audio_id], vec![4, 1, 2, 3, 2]); +} + +/// The `(video, audio)` track ids declared by an init segment. +fn track_ids(init: &Bytes) -> (u32, u32) { + let mut cursor = Cursor::new(init.as_ref()); + let mut video = None; + let mut audio = None; + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).expect("decode init") { + let mp4_atom::Any::Moov(moov) = atom else { + continue; + }; + for trak in &moov.trak { + match trak.mdia.hdlr.handler.to_string().as_str() { + "vide" => video = Some(trak.tkhd.track_id), + "soun" => audio = Some(trak.tkhd.track_id), + other => panic!("unexpected handler {other}"), + } + } + } + (video.expect("a video trak"), audio.expect("an audio trak")) +} + +/// The `(track_id, sample count)` of every `traf` in a media fragment. +fn traf_samples(fragment: &Bytes) -> Vec<(u32, usize)> { + let mut cursor = Cursor::new(fragment.as_ref()); + let mut trafs = Vec::new(); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).expect("decode fragment") { + let mp4_atom::Any::Moof(moof) = atom else { + continue; + }; + for traf in moof.traf { + let samples = traf.trun.iter().map(|trun| trun.entries.len()).sum(); + trafs.push((traf.tfhd.track_id, samples)); + } + } + trafs +} + +/// Every fragment the exporter can produce without waiting for more input. +async fn drain_now( + exporter: &mut crate::container::fmp4::Export, +) -> Vec { + let mut fragments = Vec::new(); + while let Ok(next) = tokio::time::timeout(std::time::Duration::from_millis(1), exporter.next_fragment()).await { + match next.expect("exporter failed") { + Some(fragment) => fragments.push(fragment), + None => break, + } + } + fragments +} + /// The next fragment, required to be ready without another frame arriving. async fn fragment_now( exporter: &mut crate::container::fmp4::Export, diff --git a/rs/moq-mux/src/container/test_util.rs b/rs/moq-mux/src/container/test_util.rs index 81012805db..c2a052143d 100644 --- a/rs/moq-mux/src/container/test_util.rs +++ b/rs/moq-mux/src/container/test_util.rs @@ -10,7 +10,7 @@ pub(crate) struct Live { pub(crate) track: crate::container::Producer, pub(crate) catalog: crate::catalog::Producer, consumer: moq_net::broadcast::Consumer, - _broadcast: moq_net::broadcast::Producer, + broadcast: moq_net::broadcast::Producer, } impl Live { @@ -27,10 +27,26 @@ impl Live { track: crate::container::Producer::new(track, crate::catalog::hang::Container::Legacy), catalog, consumer, - _broadcast: broadcast, + broadcast, } } + /// Add another track named `name` to the same broadcast, with `insert` + /// registering its catalog rendition. Used to build an A/V broadcast. + pub(crate) fn add_track( + &mut self, + name: &str, + insert: impl FnOnce(&mut crate::catalog::Producer, String), + ) -> crate::container::Producer { + let name = self.broadcast.unique_name(name); + let track = self + .broadcast + .create_track(name, hang::container::track_info()) + .unwrap(); + insert(&mut self.catalog, track.name().to_string()); + crate::container::Producer::new(track, crate::catalog::hang::Container::Legacy) + } + /// One Avc3-shape H.264 rendition (320x240 at 30 fps). pub(crate) fn avc3() -> Self { Self::new(".avc3", |catalog, name| { From e80cb3b171574a212d8dc526394e50ecc21b91a3 Mon Sep 17 00:00:00 2001 From: Frando Date: Wed, 2 Sep 2026 04:54:20 +0200 Subject: [PATCH 2/6] fix(moq-mux): write an ended track's tail without waiting out the others An export whose video ends while the audio plays on held the video's last GOP back until the audio ended too. Audio falls back to per-frame fragments once no unfinished video track is left to roll it, so a source with another packet always ready (a recording, a fetch) hands back a fragment on every poll and the exporter never reaches the idle step that drains a finished track's buffer. The video tail then landed after the whole rest of the audio, a tfdt inversion as long as the audio outlived the video. Write a finished track's buffered tail as soon as nothing another track holds starts earlier, which is where it belongs in the file, rather than leave it for a moment when every track happens to be idle at once. The tests now assert that a fragment's tfdt never goes backwards, which is what the inversion broke. (cherry picked from commit 59d9d5196cb3a729d709c3dd19173fb05693c998) --- rs/moq-mux/src/container/fmp4/export.rs | 58 +++++++++- rs/moq-mux/src/container/fmp4/export_test.rs | 115 ++++++++++++++++++- 2 files changed, 166 insertions(+), 7 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index 1b466437d4..b35f8b4e9e 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -111,6 +111,21 @@ struct Fmp4Track { sequence_number: u32, } +impl Fmp4Track { + /// When the next content this track will contribute starts: the first frame + /// of its buffered run, or its pending frame when the buffer is empty. + /// + /// A [`Duration`] rather than a [`Timestamp`], whose ordering breaks a + /// cross-scale tie by scale, so a 48 kHz audio track would sort ahead of a + /// 90 kHz video one at t=0. + fn next_start(&self) -> Option { + self.buffer + .first() + .or(self.pending.as_ref()) + .map(|frame| Duration::from(frame.timestamp)) + } +} + impl Export { /// Subscribe to `source` and produce fMP4 byte chunks, driving track /// (un)subscription from `catalog`. @@ -264,7 +279,19 @@ impl Export { return Poll::Pending; } - // 4. Pick the track whose pending frame has the smallest timestamp and + // 4. Write out the buffered tail of a track that has ended, as soon as no + // other track still holds something earlier. Step 6 does the same, but + // only once every track is idle at the same moment, which a track fed by + // a recording or a fetch never is: the tail of a video track that ends + // mid-broadcast would trail the whole rest of the audio. + if let Some(name) = self.stranded_tail() { + let track = self.tracks.get_mut(&name).unwrap(); + let frames = std::mem::take(&mut track.buffer); + let fragment = emit_fragment(track, frames, None)?; + return Poll::Ready(Ok(Some(fragment))); + } + + // 5. Pick the track whose pending frame has the smallest timestamp and // decide whether to flush its buffer before appending the new frame. let chosen = self .tracks @@ -331,8 +358,10 @@ impl Export { continue; } - // 5. No pending frames. Flush any finished tracks' remaining buffers, - // in ascending first-frame-timestamp order. + // 6. Nothing pending anywhere. Step 4 has already written every finished + // tail that was next in order, so what's left is one held back by a track + // that stalled without ending. Write it anyway, earliest first, rather + // than wait out a track that may never speak again. let flushable = self .tracks .iter() @@ -353,12 +382,12 @@ impl Export { return Poll::Ready(Ok(Some(fragment))); } - // 6. If catalog is closed and every track is finished and drained, we're done. + // 7. If catalog is closed and every track is finished and drained, we're done. if self.catalog.is_none() && self.tracks.values().all(|t| t.finished && t.buffer.is_empty()) { return Poll::Ready(Ok(None)); } - // 7. Drop finished tracks with empty buffers so the next catalog update can re-add a track of the same name. + // 8. Drop finished tracks with empty buffers so the next catalog update can re-add a track of the same name. self.tracks .retain(|_, t| !(t.finished && t.pending.is_none() && t.buffer.is_empty())); @@ -366,6 +395,25 @@ impl Export { } } + /// The track whose ended run can be written now: it is finished, so nothing + /// more will join its buffer, and nothing another track holds starts earlier. + /// + /// A track that ends mid-broadcast has no successor to flush its tail. Video + /// that has ended no longer draws the GOP boundary that rolls the audio with + /// it either, so the audio switches to per-frame fragments and keeps the + /// exporter busy, which is what strands the tail: it would otherwise wait for + /// every other track to end too. + fn stranded_tail(&self) -> Option { + let earliest = self.tracks.values().filter_map(Fmp4Track::next_start).min()?; + self.tracks + .iter() + .filter(|(_, track)| track.finished && !track.buffer.is_empty()) + .map(|(name, track)| (Duration::from(track.buffer[0].timestamp), !track.is_video, name)) + .filter(|(start, _, _)| *start <= earliest) + .min() + .map(|(_, _, name)| name.clone()) + } + /// Encode every audio track's buffered run as a fragment, tagged with the /// timestamp of its first sample so the caller can order it against the /// video fragment it accompanies. diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index d94f150bb4..67dfa7dc6c 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -806,9 +806,11 @@ async fn audio_fragments_roll_with_the_video_gop() { assert!(init.init); let (video_id, audio_id) = track_ids(&init.data); + let fragments = drain_now(&mut exporter).await; + assert_ascending_starts(&init.data, &fragments); + // Every fragment, tagged with the track it belongs to and its sample count. - let fragments: Vec<(u32, usize)> = drain_now(&mut exporter) - .await + let fragments: Vec<(u32, usize)> = fragments .iter() .map(|fragment| { let trafs = traf_samples(&fragment.data); @@ -832,6 +834,71 @@ async fn audio_fragments_roll_with_the_video_gop() { ); } +/// A video track that ends while the audio plays on must not leave its last GOP +/// behind the whole rest of the audio. +/// +/// Audio falls back to per-frame fragmenting once no unfinished video track is +/// left to roll it, so a source with another packet always ready (a recording, a +/// fetch) keeps producing fragments and the exporter never reaches the idle path +/// that drains a finished track's buffer. The video's tail then lands after every +/// audio fragment, however long the audio runs on: a `tfdt` inversion with no +/// bound on it. +#[tokio::test(start_paused = true)] +async fn video_that_ends_first_writes_its_tail_in_order() { + use hang::catalog::{AudioCodec, AudioConfig, Container}; + + let mut live = Live::avc3(); + let mut audio = live.add_track(".opus", |catalog, name| { + let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); + config.container = Container::Legacy; + catalog.lock().audio.renditions.insert(name, config); + }); + + // Two 30 fps GOPs of three frames each, and then the video ends. + for i in 0..6u64 { + live.track.write(video_frame(i * 33_000, i % 3 == 0)).unwrap(); + } + live.track.finish().unwrap(); + // The audio runs on for another half second, always ready. + for i in 0..30u64 { + audio + .write(raw_frame(i * 20_000, &[0x08, 0xaa, 0xbb, 0xcc], true)) + .unwrap(); + } + + let mut exporter = crate::container::fmp4::Export::new(live.source(), live.catalog_stream().await); + let init = fragment_now(&mut exporter).await; + assert!(init.init); + let (video_id, audio_id) = track_ids(&init.data); + + let fragments = drain_now(&mut exporter).await; + assert_ascending_starts(&init.data, &fragments); + + let counts: Vec<(u32, usize)> = fragments + .iter() + .map(|fragment| { + let trafs = traf_samples(&fragment.data); + assert_eq!(trafs.len(), 1, "expected one traf per fragment"); + trafs[0] + }) + .collect(); + + // The first GOP rolls the first five packets with it; the second GOP is the + // video tail, which goes out as soon as the video ends rather than waiting + // for the audio, followed by the audio buffered up to that point. + assert_eq!( + counts[..4], + [(video_id, 3), (audio_id, 5), (video_id, 3), (audio_id, 4)], + "the video tail must not trail the audio", + ); + // Everything after it is audio, one fragment per packet. + assert!(counts[4..].iter().all(|(id, samples)| *id == audio_id && *samples == 1)); + let audio_samples: usize = counts.iter().filter(|(id, _)| *id == audio_id).map(|(_, n)| n).sum(); + assert_eq!(audio_samples, 30, "every audio packet must be written exactly once"); + let video_samples: usize = counts.iter().filter(|(id, _)| *id == video_id).map(|(_, n)| n).sum(); + assert_eq!(video_samples, 6, "every video frame must be written exactly once"); +} + /// A simulcast ladder rolls each video track on its own keyframes: one /// rendition's GOP boundary must not split another's, or the second rendition's /// fragments stop being independently decodable. Audio rolls at whichever @@ -942,6 +1009,50 @@ fn traf_samples(fragment: &Bytes) -> Vec<(u32, usize)> { trafs } +/// The timescale the init segment declares for each track id. +fn track_timescales(init: &Bytes) -> std::collections::BTreeMap { + let mut cursor = Cursor::new(init.as_ref()); + let mut scales = std::collections::BTreeMap::new(); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).expect("decode init") { + if let mp4_atom::Any::Moov(moov) = atom { + for trak in &moov.trak { + scales.insert(trak.tkhd.track_id, u64::from(trak.mdia.mdhd.timescale)); + } + } + } + scales +} + +/// The presentation time each fragment starts at (its `tfdt`), in seconds, read +/// at the timescale its track declares in the init segment. +fn fragment_starts(init: &Bytes, fragments: &[crate::container::fmp4::Fragment]) -> Vec<(u32, f64)> { + let scales = track_timescales(init); + fragments + .iter() + .map(|fragment| { + let traf = super::first_traf(&fragment.data); + let track_id = traf.tfhd.track_id; + let tfdt = traf.tfdt.expect("a tfdt").base_media_decode_time; + let scale = *scales.get(&track_id).expect("a track in the init segment"); + (track_id, tfdt as f64 / scale as f64) + }) + .collect() +} + +/// Assert the fragments' start times never go backwards, across every track. +fn assert_ascending_starts(init: &Bytes, fragments: &[crate::container::fmp4::Fragment]) { + let starts = fragment_starts(init, fragments); + for pair in starts.windows(2) { + let [(previous_id, previous), (id, start)] = pair else { + unreachable!(); + }; + assert!( + start >= previous, + "track {id} starts at {start}s after track {previous_id} started at {previous}s: {starts:?}", + ); + } +} + /// Every fragment the exporter can produce without waiting for more input. async fn drain_now( exporter: &mut crate::container::fmp4::Export, From 9dcb0cdcfd476f4d13025596fedf8da22089201e Mon Sep 17 00:00:00 2001 From: Frando Date: Wed, 2 Sep 2026 04:58:51 +0200 Subject: [PATCH 3/6] fix(moq-mux): bound the fMP4 audio buffer when no video keyframe rolls it Audio rolls its fragment at a video GOP boundary, which leaves it with no trigger of its own: a video track that stops delivering without ending, or an encoder whose keyframe interval is effectively infinite (screen sharing), never draws another boundary, and the audio buffer grows for the length of the session. That is the failure rolling at the GOP set out to remove, and `irl record` has no cap of its own to fall back on. Roll an audio run that spans five seconds regardless. The value sits above any keyframe interval a broadcast is likely to use (one to four seconds), so a healthy stream never reaches it and its audio fragments still cover exactly the spans its video fragments do; a stall now costs one fragment per five seconds of audio, around 80 KB per track at 128 kb/s, instead of the whole session. An explicit `with_fragment_duration` still takes precedence, and a video track keeps rolling on its own keyframes alone. (cherry picked from commit a018fbb08f49fe8b1cc7280d2c22ca25b8970595) --- rs/moq-mux/src/container/fmp4/export.rs | 42 +++++++++++---- rs/moq-mux/src/container/fmp4/export_test.rs | 56 ++++++++++++++++++++ 2 files changed, 88 insertions(+), 10 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index b35f8b4e9e..e1281452c7 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -85,7 +85,8 @@ struct Fmp4Track { /// Frames accumulated for the current fragment. Flushed as a single /// moof+mdat at a GOP boundary (a keyframe on this track for video, the - /// keyframe of any video track for audio) or on the duration cap. + /// keyframe of any video track for audio), on the duration cap, or, for + /// audio, once the run spans [`MAX_AUDIO_SPAN`]. buffer: Vec, /// Whether the first frame of the current `buffer` was a keyframe, i.e. the @@ -158,12 +159,13 @@ impl Export { /// Cap the fragment (moof+mdat) duration. /// /// By default video fragments roll over on each keyframe (one fragment per - /// GOP) and audio rolls with them; a broadcast with no video at all emits - /// one audio fragment per sample, having no GOP to follow. Setting this + /// GOP) and audio rolls with them, up to a five second backstop for when the + /// video stops drawing boundaries; a broadcast with no video at all + /// emits one audio fragment per sample, having no GOP to follow. Setting this /// caps each fragment to roughly `duration` of frames, useful for /// downstream consumers that throttle by fragment rate. [`Duration::ZERO`] /// emits one fragment per frame (the historical behavior); otherwise the - /// cap applies in addition to GOP rollover. + /// cap applies in addition to GOP rollover, and replaces the audio backstop. /// /// Accepts either `Duration` or `Option` (where `None` restores /// the per-GOP default). @@ -661,11 +663,27 @@ pub(crate) fn extract_init( Ok(()) } +/// How much presentation time an audio track's buffered run may span before it +/// rolls a fragment of its own, when no explicit +/// [`fragment_duration`](Export::with_fragment_duration) is set. +/// +/// Audio is otherwise rolled by a video keyframe, and a video track that stops +/// delivering without ending, or an encoder whose keyframe interval is +/// effectively infinite (screen sharing), never draws that boundary again. The +/// cap is what stops the buffer growing for the length of the session. +/// +/// Five seconds sits above any keyframe interval a broadcast is likely to use +/// (one to four seconds), so a healthy stream never reaches it and its audio +/// fragments still cover exactly the spans its video fragments do. What it costs +/// when a stall does reach it is one fragment per five seconds of audio, holding +/// around 80 KB per track at 128 kb/s. +const MAX_AUDIO_SPAN: Duration = Duration::from_secs(5); + /// Should we flush `track.buffer` before appending the incoming `frame`? -/// Triggers on a video keyframe (one fragment per GOP) or the duration cap. -/// Audio has no keyframe of its own and is rolled by the video track instead, -/// via [`Export::flush_audio`]. Per-frame modes never buffer and are handled -/// before this check. +/// Triggers on a video keyframe (one fragment per GOP), the duration cap, or +/// [`MAX_AUDIO_SPAN`]. Audio has no keyframe of its own and is rolled by the +/// video track instead, via [`Export::flush_audio`]. Per-frame modes never +/// buffer and are handled before this check. fn should_flush(track: &Fmp4Track, frame: &Frame, fragment_duration: Option) -> bool { if track.buffer.is_empty() { return false; @@ -673,8 +691,12 @@ fn should_flush(track: &Fmp4Track, frame: &Frame, fragment_duration: Option cap, + // A video track rolls on its own keyframes; audio depends on someone + // else's and needs a bound of its own for when none arrives. + None if track.is_video => return false, + None => MAX_AUDIO_SPAN, }; // Frames within a track are in *decode* order; B-frames have non-monotonic // PTS, so the span of the buffer is min..max of all PTS. diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index 67dfa7dc6c..746f52e0a8 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -899,6 +899,62 @@ async fn video_that_ends_first_writes_its_tail_in_order() { assert_eq!(video_samples, 6, "every video frame must be written exactly once"); } +/// Audio must not buffer without bound while a video track stalls. +/// +/// Rolling the audio at the video GOP boundary leaves it with no trigger of its +/// own, and a video track that stops delivering without ending, or an encoder +/// with an effectively infinite keyframe interval (screen sharing), never draws +/// another boundary. The audio buffer would then grow for the length of the +/// session, which is the failure the GOP rollover set out to remove. +#[tokio::test(start_paused = true)] +async fn audio_rolls_on_its_own_cap_when_the_video_stalls() { + use hang::catalog::{AudioCodec, AudioConfig, Container}; + + let mut live = Live::avc3(); + let mut audio = live.add_track(".opus", |catalog, name| { + let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); + config.container = Container::Legacy; + catalog.lock().audio.renditions.insert(name, config); + }); + + // One keyframe, enough to build the init segment, and then the video stalls: + // it delivers nothing more and never ends. + live.track.write(video_frame(0, true)).unwrap(); + // Twelve seconds of 20 ms Opus packets. + for i in 0..600u64 { + audio + .write(raw_frame(i * 20_000, &[0x08, 0xaa, 0xbb, 0xcc], true)) + .unwrap(); + } + + let mut exporter = crate::container::fmp4::Export::new(live.source(), live.catalog_stream().await); + let init = fragment_now(&mut exporter).await; + assert!(init.init); + let audio_id = track_ids(&init.data).1; + + let fragments = drain_now(&mut exporter).await; + assert_ascending_starts(&init.data, &fragments); + for fragment in &fragments { + assert!( + fragment.duration <= 5.0, + "an audio fragment spanned {} seconds with no video to roll it", + fragment.duration, + ); + } + + // Two full runs of the cap; the last two seconds stay buffered, still able to + // roll with the video if it ever comes back. + let counts: Vec<(u32, usize)> = fragments + .iter() + .map(|fragment| { + let trafs = traf_samples(&fragment.data); + assert_eq!(trafs.len(), 1, "expected one traf per fragment"); + trafs[0] + }) + .collect(); + assert_eq!(counts, vec![(audio_id, 250), (audio_id, 250)]); +} + /// A simulcast ladder rolls each video track on its own keyframes: one /// rendition's GOP boundary must not split another's, or the second rendition's /// fragments stop being independently decodable. Audio rolls at whichever From 787d6315e176a6e60542d54198dd3e7b2380af6d Mon Sep 17 00:00:00 2001 From: Frando Date: Wed, 2 Sep 2026 05:07:31 +0200 Subject: [PATCH 4/6] fix(moq-mux): number and order a GOP boundary's fragments as the file has them A video keyframe rolls a fragment on every track, and the exporter encoded the video's run first, then the audio's, and sorted the results by first sample before handing them back. Two things came out of order. `mfhd.sequence_number` counted per track, so a two-track file went 1,1,2,2,3,3 and a simulcast ladder stepped backwards outright, where ISO/IEC 14496-12 section 8.8.5 wants the number to increase from one movie fragment to the next. The sort also compared timestamps, whose ordering settles a cross-scale tie by scale, so 48 kHz audio sorted ahead of 90 kHz video at t=0: the one tie the "video first" rule exists for. Only a CMAF source holds two scales at once, the legacy wire format normalizing everything to microseconds, so no test covers that half. Decide the order first and encode in it, drawing sequence numbers from a single counter for the file. Each track's own numbers still ascend within the shared sequence, which is what CMAF asks for. (cherry picked from commit f236022d032801762170d276f53a2a8e36c970b1) --- rs/moq-mux/src/container/fmp4/export.rs | 127 ++++++++++++------- rs/moq-mux/src/container/fmp4/export_test.rs | 34 ++++- 2 files changed, 111 insertions(+), 50 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index e1281452c7..da105192ab 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -56,6 +56,15 @@ pub struct Export { /// audio track alongside the video one, which produces more than one fragment /// at a time; they're returned in ascending first-sample order. outgoing: VecDeque, + + /// The `mfhd.sequence_number` the next fragment gets. + /// + /// One counter for the file rather than one per track: ISO/IEC 14496-12 + /// section 8.8.5 wants the numbers to increase from one movie fragment to the + /// next, and a per-track counter written into a multi-track file steps + /// backwards on every other fragment. Each track's own numbers still ascend + /// within the shared sequence, which is what CMAF asks for. + sequence_number: u32, } /// One emitted CMAF chunk: either the init segment or a moof+mdat fragment, @@ -109,7 +118,6 @@ struct Fmp4Track { track_id: u32, timescale: u64, - sequence_number: u32, } impl Fmp4Track { @@ -144,6 +152,7 @@ impl Export { catalog_snapshot: None, init_emitted: false, outgoing: VecDeque::new(), + sequence_number: 1, } } @@ -289,7 +298,7 @@ impl Export { if let Some(name) = self.stranded_tail() { let track = self.tracks.get_mut(&name).unwrap(); let frames = std::mem::take(&mut track.buffer); - let fragment = emit_fragment(track, frames, None)?; + let fragment = emit_fragment(track, &mut self.sequence_number, frames, None)?; return Poll::Ready(Ok(Some(fragment))); } @@ -321,36 +330,25 @@ impl Export { // first and retry this frame on the next poll. if !track.buffer.is_empty() { let frames = std::mem::take(&mut track.buffer); - let fragment = emit_fragment(track, frames, Some(&frame))?; + let fragment = emit_fragment(track, &mut self.sequence_number, frames, Some(&frame))?; track.pending = Some(frame); return Poll::Ready(Ok(Some(fragment))); } track.buffer_independent = frame.keyframe; - let fragment = emit_fragment(track, vec![frame], None)?; + let fragment = emit_fragment(track, &mut self.sequence_number, vec![frame], None)?; return Poll::Ready(Ok(Some(fragment))); } if should_flush(track, &frame, frag) { - // A video keyframe closes a GOP for the whole presentation, not - // just for this track: audio has no keyframes of its own, so it - // rolls here to keep every track's fragments covering the same - // span. Without it an audio track would buffer until it finished. - let gop = track.is_video && frame.keyframe; - let start = track.buffer[0].timestamp; + if track.is_video && frame.keyframe { + self.roll_gop(&name, frame)?; + return Poll::Ready(Ok(self.outgoing.pop_front())); + } let frames = std::mem::take(&mut track.buffer); - let fragment = emit_fragment(track, frames, Some(&frame))?; + let fragment = emit_fragment(track, &mut self.sequence_number, frames, Some(&frame))?; // The flushed run is done; the incoming frame opens the next buffer. track.buffer_independent = frame.keyframe; track.buffer.push(frame); - if !gop { - return Poll::Ready(Ok(Some(fragment))); - } - // Video first, so a tie at the same start puts the track that - // drove the boundary ahead of the audio that followed it. - let mut rolled = vec![(start, fragment)]; - rolled.extend(self.flush_audio()?); - rolled.sort_by_key(|(start, _)| *start); - self.outgoing.extend(rolled.into_iter().map(|(_, fragment)| fragment)); - return Poll::Ready(Ok(self.outgoing.pop_front())); + return Poll::Ready(Ok(Some(fragment))); } if track.buffer.is_empty() { track.buffer_independent = frame.keyframe; @@ -380,7 +378,7 @@ impl Export { if let Some(name) = flushable { let track = self.tracks.get_mut(&name).unwrap(); let frames = std::mem::take(&mut track.buffer); - let fragment = emit_fragment(track, frames, None)?; + let fragment = emit_fragment(track, &mut self.sequence_number, frames, None)?; return Poll::Ready(Ok(Some(fragment))); } @@ -416,28 +414,55 @@ impl Export { .map(|(_, _, name)| name.clone()) } - /// Encode every audio track's buffered run as a fragment, tagged with the - /// timestamp of its first sample so the caller can order it against the - /// video fragment it accompanies. + /// Roll the run that `frame`, a keyframe, has just closed on video track + /// `name`, along with every audio track's buffered run, into + /// [`Self::outgoing`]. `frame` opens the video track's next run. /// - /// Called only from a video GOP boundary, so no audio track is the one - /// currently being appended to. - fn flush_audio(&mut self) -> Result> { - let mut fragments = Vec::new(); - for track in self.tracks.values_mut() { - if track.is_video || track.buffer.is_empty() { - continue; + /// A keyframe closes a GOP for the whole presentation, not just for the track + /// it arrived on: audio has no keyframes of its own, so it rolls here to keep + /// every track's fragments covering the same span. Without it an audio track + /// would buffer until it finished. + /// + /// The runs are encoded in ascending first-sample order, so the fragments are + /// written in the order they leave in and their `mfhd` sequence numbers + /// ascend with the file. + fn roll_gop(&mut self, name: &str, frame: Frame) -> Result<()> { + let track = self.tracks.get_mut(name).unwrap(); + let start = Duration::from(track.buffer[0].timestamp); + let mut closed = std::mem::take(&mut track.buffer); + + // Video leads the audio it rolled when both start at the same instant. + // Ordered by elapsed time rather than by `Timestamp`, whose ordering + // settles a cross-scale tie by scale: 48 kHz audio would otherwise sort + // ahead of 90 kHz video at t=0, which is the tie this exists to settle. + let mut rolled = vec![(start, false, name.to_string())]; + for (name, track) in &self.tracks { + if !track.is_video && !track.buffer.is_empty() { + rolled.push((Duration::from(track.buffer[0].timestamp), true, name.clone())); } - let start = track.buffer[0].timestamp; - let frames = std::mem::take(&mut track.buffer); - // The pending frame is this track's next sample, so it bounds the - // duration of the run's last one. - let successor = track.pending.take(); - let fragment = emit_fragment(track, frames, successor.as_ref())?; - track.pending = successor; - fragments.push((start, fragment)); } - Ok(fragments) + rolled.sort(); + + for (_, audio, name) in rolled { + let track = self.tracks.get_mut(&name).unwrap(); + // Whatever the track's next sample is bounds the duration of this + // run's last one: the keyframe for the video track that drew the + // boundary, the pending frame for an audio track. + let (frames, successor) = if audio { + (std::mem::take(&mut track.buffer), track.pending.clone()) + } else { + (std::mem::take(&mut closed), Some(frame.clone())) + }; + let fragment = emit_fragment(track, &mut self.sequence_number, frames, successor.as_ref())?; + self.outgoing.push_back(fragment); + } + + // The closed run is written; the keyframe opens the video track's next one. + let track = self.tracks.get_mut(name).unwrap(); + track.buffer_independent = frame.keyframe; + track.buffer.push(frame); + + Ok(()) } fn update_catalog(&mut self, catalog: &Catalog) -> Result<()> { @@ -488,7 +513,6 @@ impl Export { finished: false, track_id: next_track_id, timescale, - sequence_number: 1, }, ); next_track_id += 1; @@ -516,7 +540,6 @@ impl Export { finished: false, track_id: next_track_id, timescale, - sequence_number: 1, }, ); next_track_id += 1; @@ -710,13 +733,14 @@ fn should_flush(track: &Fmp4Track, frame: &Frame, fragment_duration: Option= cap } -/// Encode a buffered run of samples as a single CMAF moof+mdat fragment. -fn encode_fragment(track: &mut Fmp4Track, frames: Vec) -> Result { +/// Encode a buffered run of samples as a single CMAF moof+mdat fragment, +/// consuming the next `mfhd` sequence number of the file. +fn encode_fragment(track: &Fmp4Track, sequence_number: &mut u32, frames: Vec) -> Result { if frames.is_empty() { return Err(Error::NoFrames.into()); } - let seq = track.sequence_number; - track.sequence_number += 1; + let seq = *sequence_number; + *sequence_number += 1; let timescale = moq_net::Timescale::new(track.timescale)?; let info = crate::container::fmp4::FragmentInfo { track_id: track.track_id, @@ -727,7 +751,12 @@ fn encode_fragment(track: &mut Fmp4Track, frames: Vec) -> Result { } /// Encode a buffered run and wrap it with the metadata a segmenting consumer needs. -fn emit_fragment(track: &mut Fmp4Track, mut frames: Vec, successor: Option<&Frame>) -> Result { +fn emit_fragment( + track: &Fmp4Track, + sequence_number: &mut u32, + mut frames: Vec, + successor: Option<&Frame>, +) -> Result { apply_codec_durations(&mut frames, track.opus); // Audio has no keyframes, so every audio fragment is independent; video is // independent only when its buffer opened on a keyframe (a GOP boundary). @@ -735,7 +764,7 @@ fn emit_fragment(track: &mut Fmp4Track, mut frames: Vec, successor: Optio let timescale = moq_net::Timescale::new(track.timescale)?; infer_missing_durations(&mut frames, successor, track.default_frame, timescale)?; let duration = fragment_seconds(&frames, track.default_frame); - let data = encode_fragment(track, frames)?; + let data = encode_fragment(track, sequence_number, frames)?; Ok(Fragment { data, init: false, diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index 746f52e0a8..09e7c25371 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -808,6 +808,7 @@ async fn audio_fragments_roll_with_the_video_gop() { let fragments = drain_now(&mut exporter).await; assert_ascending_starts(&init.data, &fragments); + assert_ascending_sequence_numbers(&fragments); // Every fragment, tagged with the track it belongs to and its sample count. let fragments: Vec<(u32, usize)> = fragments @@ -873,6 +874,7 @@ async fn video_that_ends_first_writes_its_tail_in_order() { let fragments = drain_now(&mut exporter).await; assert_ascending_starts(&init.data, &fragments); + assert_ascending_sequence_numbers(&fragments); let counts: Vec<(u32, usize)> = fragments .iter() @@ -934,6 +936,7 @@ async fn audio_rolls_on_its_own_cap_when_the_video_stalls() { let fragments = drain_now(&mut exporter).await; assert_ascending_starts(&init.data, &fragments); + assert_ascending_sequence_numbers(&fragments); for fragment in &fragments { assert!( fragment.duration <= 5.0, @@ -1005,8 +1008,11 @@ async fn simulcast_keyframes_only_roll_their_own_video_track() { assert!(init.init); let audio_id = track_ids(&init.data).1; + let fragments = drain_now(&mut exporter).await; + assert_ascending_sequence_numbers(&fragments); + let mut per_track: std::collections::BTreeMap> = std::collections::BTreeMap::new(); - for fragment in drain_now(&mut exporter).await { + for fragment in fragments { let trafs = traf_samples(&fragment.data); assert_eq!(trafs.len(), 1, "expected one traf per fragment"); let (track_id, samples) = trafs[0]; @@ -1109,6 +1115,32 @@ fn assert_ascending_starts(init: &Bytes, fragments: &[crate::container::fmp4::Fr } } +/// The `mfhd` sequence numbers must ascend from one fragment to the next, which +/// is what ISO/IEC 14496-12 section 8.8.5 asks of a file. +fn assert_ascending_sequence_numbers(fragments: &[crate::container::fmp4::Fragment]) { + let numbers: Vec = fragments + .iter() + .map(|fragment| first_moof(&fragment.data).mfhd.sequence_number) + .collect(); + for pair in numbers.windows(2) { + assert!( + pair[1] > pair[0], + "sequence numbers must ascend in file order, got {numbers:?}", + ); + } +} + +/// The first `moof` of a media fragment. +fn first_moof(fragment: &Bytes) -> mp4_atom::Moof { + let mut cursor = Cursor::new(fragment.as_ref()); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).expect("decode fragment") { + if let mp4_atom::Any::Moof(moof) = atom { + return moof; + } + } + panic!("a fragment with no moof"); +} + /// Every fragment the exporter can produce without waiting for more input. async fn drain_now( exporter: &mut crate::container::fmp4::Export, From 2cbdab6043017dbb16a9fce22b571250dc8057e0 Mon Sep 17 00:00:00 2001 From: Frando Date: Wed, 2 Sep 2026 05:09:33 +0200 Subject: [PATCH 5/6] docs(moq-mux): say what the fMP4 export really orders `Export::next` claimed fragments arrive in ascending timestamp order across tracks. They don't: each video rendition rolls on its own keyframes, so a simulcast ladder interleaves the renditions out of order, by 14 ms in the test here and by seconds between a 10 second GOP and a 1 second one. Audio rolled by its own cap has no video fragment beside it either. Holding fragments back to sort them would cost the length of the longest GOP in the ladder in added latency, which a live export should not pay for an ordering no fMP4 consumer needs: a demuxer indexes by track and reads tfdt, and a segmenting consumer is already told to narrow the catalog to one rendition, where the order does hold. So state the guarantee that holds instead: every track ascends in itself, and a GOP boundary writes what it rolls in order, which covers a broadcast with a single video rendition. The simulcast test now asserts exactly that. (cherry picked from commit b44a6edad3ca2ba342719f330566f2d42ca3213a) --- rs/moq-mux/src/container/fmp4/export.rs | 22 +++++++++++++++----- rs/moq-mux/src/container/fmp4/export_test.rs | 19 +++++++++++++++++ 2 files changed, 36 insertions(+), 5 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index da105192ab..b4f0addc89 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -18,8 +18,9 @@ use moq_net::Timestamp; /// Built from a [`Source`](crate::Source), `Export` subscribes to the hang catalog, /// (un)subscribes per-rendition tracks as the catalog changes, decodes both Legacy and /// CMAF tracks via a per-track source, and re-encodes everything as a merged init -/// segment + moof+mdat fragments in presentation-timestamp order across tracks. This -/// is what an fMP4 player (e.g. ffplay, MSE) expects. +/// segment + moof+mdat fragments, each track in presentation-timestamp order (see +/// [`next`](Self::next) for how far that goes across tracks). This is what an fMP4 +/// player (e.g. ffplay, MSE) expects. /// /// Use [`next`](Self::next) to pull byte chunks: the first call returns the merged /// init segment (ftyp + multi-track moov), subsequent calls return moof+mdat @@ -186,9 +187,20 @@ impl Export { /// Get the next byte chunk. /// /// The first call returns the merged init segment (ftyp + multi-track moov); each - /// subsequent call returns one moof+mdat fragment. Fragments arrive in ascending - /// timestamp order across tracks. Returns `None` when the catalog and every track - /// have ended. + /// subsequent call returns one moof+mdat fragment. Returns `None` when the catalog + /// and every track have ended. + /// + /// A track's own fragments always ascend in timestamp, and a GOP boundary writes + /// the fragments it rolls in ascending order, so a broadcast with one video + /// rendition comes out ordered across its tracks too, for as long as the video + /// keeps drawing boundaries. Beyond that the order is only approximate: each + /// rendition of a simulcast ladder rolls on its own keyframes, which need not + /// coincide, and audio rolled by its own cap has no video fragment beside it, so a + /// fragment can follow one that starts later, by as much as the difference between + /// the GOP lengths. Putting that right would mean holding every fragment for the + /// length of the longest GOP in the ladder, which is not a trade a live export + /// should make; narrow the catalog to a single rendition with + /// [`Stream::select`](crate::catalog::Stream::select) where the order matters. pub async fn next(&mut self) -> Result> { Ok(self.next_fragment().await?.map(|f| f.data)) } diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index 09e7c25371..d544ce59b2 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -1010,6 +1010,10 @@ async fn simulcast_keyframes_only_roll_their_own_video_track() { let fragments = drain_now(&mut exporter).await; assert_ascending_sequence_numbers(&fragments); + // Only per track: two renditions rolling on keyframes that don't coincide + // interleave their fragments out of order, which is what `Export::next` + // promises and all it promises. + assert_each_track_ascends(&init.data, &fragments); let mut per_track: std::collections::BTreeMap> = std::collections::BTreeMap::new(); for fragment in fragments { @@ -1115,6 +1119,21 @@ fn assert_ascending_starts(init: &Bytes, fragments: &[crate::container::fmp4::Fr } } +/// Every track's own fragments must ascend in start time, whatever the order +/// they are interleaved in. +fn assert_each_track_ascends(init: &Bytes, fragments: &[crate::container::fmp4::Fragment]) { + let starts = fragment_starts(init, fragments); + let mut previous: std::collections::BTreeMap = std::collections::BTreeMap::new(); + for (id, start) in &starts { + if let Some(previous) = previous.insert(*id, *start) { + assert!( + *start > previous, + "track {id} starts at {start}s after starting at {previous}s: {starts:?}", + ); + } + } +} + /// The `mfhd` sequence numbers must ascend from one fragment to the next, which /// is what ISO/IEC 14496-12 section 8.8.5 asks of a file. fn assert_ascending_sequence_numbers(fragments: &[crate::container::fmp4::Fragment]) { From e0f9951852eafd7ac5a6c15c398ed9012b7b5955 Mon Sep 17 00:00:00 2001 From: Frando Date: Thu, 10 Sep 2026 09:46:42 -0400 Subject: [PATCH 6/6] fix(moq-mux): keep an audio sample on the keyframe in the GOP it opens A fragment covers up to the boundary and not including it, and rolling took the whole audio buffer, so a packet whose timestamp equalled the keyframe's went into the run that keyframe closed. That fragment then overlapped the boundary and the next video fragment began with no audio beside it until the packet after. Which way it fell depended on the order the caller wrote the two tracks in, which is not something the file should record. The buffer is split at the boundary instead, and the first sample held back bounds the duration of the last one written, which is a real successor rather than the pending frame. Also: the doc on `should_flush` pointed at an `Export::flush_audio` that does not exist; the rollover is in `roll_gop`. Both reported by review on #3358. Co-Authored-By: Claude Opus 5 (1M context) --- rs/moq-mux/src/container/fmp4/export.rs | 27 ++++++++- rs/moq-mux/src/container/fmp4/export_test.rs | 62 ++++++++++++++++++++ 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/rs/moq-mux/src/container/fmp4/export.rs b/rs/moq-mux/src/container/fmp4/export.rs index b4f0addc89..8252981b49 100644 --- a/rs/moq-mux/src/container/fmp4/export.rs +++ b/rs/moq-mux/src/container/fmp4/export.rs @@ -455,16 +455,37 @@ impl Export { } rolled.sort(); + // The instant the keyframe draws. A fragment covers up to it and not + // including it, so an audio sample landing exactly here opens the next + // run rather than closing this one. Whether such a sample had already + // been buffered depended on the order the caller wrote the two tracks + // in, which is not something the file should record. + let boundary = Duration::from(frame.timestamp); + for (_, audio, name) in rolled { let track = self.tracks.get_mut(&name).unwrap(); // Whatever the track's next sample is bounds the duration of this // run's last one: the keyframe for the video track that drew the - // boundary, the pending frame for an audio track. + // boundary, and for audio the first sample held back, or the pending + // frame when none was. let (frames, successor) = if audio { - (std::mem::take(&mut track.buffer), track.pending.clone()) + let split = track + .buffer + .iter() + .position(|buffered| Duration::from(buffered.timestamp) >= boundary) + .unwrap_or(track.buffer.len()); + let held = track.buffer.split_off(split); + let frames = std::mem::replace(&mut track.buffer, held); + let successor = track.buffer.first().cloned().or_else(|| track.pending.clone()); + (frames, successor) } else { (std::mem::take(&mut closed), Some(frame.clone())) }; + // Every sample sat at or past the boundary, so this run has nothing + // to write and the whole buffer belongs to the next one. + if frames.is_empty() { + continue; + } let fragment = emit_fragment(track, &mut self.sequence_number, frames, successor.as_ref())?; self.outgoing.push_back(fragment); } @@ -717,7 +738,7 @@ const MAX_AUDIO_SPAN: Duration = Duration::from_secs(5); /// Should we flush `track.buffer` before appending the incoming `frame`? /// Triggers on a video keyframe (one fragment per GOP), the duration cap, or /// [`MAX_AUDIO_SPAN`]. Audio has no keyframe of its own and is rolled by the -/// video track instead, via [`Export::flush_audio`]. Per-frame modes never +/// video track instead, in [`Export::roll_gop`]. Per-frame modes never /// buffer and are handled before this check. fn should_flush(track: &Fmp4Track, frame: &Frame, fragment_duration: Option) -> bool { if track.buffer.is_empty() { diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index d544ce59b2..7bb1e2a916 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -835,6 +835,68 @@ async fn audio_fragments_roll_with_the_video_gop() { ); } +/// An audio sample landing exactly on a keyframe belongs to the GOP that +/// keyframe opens, not the one it closes. +/// +/// A fragment covers up to the boundary and not including it. Draining the whole +/// audio buffer put an equal-timestamped sample in the run before it, so that +/// fragment overlapped the boundary and the next video fragment began with no +/// audio beside it until the following packet. Which way it fell depended on the +/// order the caller happened to write the two tracks in. +#[tokio::test] +async fn audio_on_the_keyframe_opens_the_next_gop() { + use hang::catalog::{AudioCodec, AudioConfig, Container}; + + let mut live = Live::avc3(); + let mut audio = live.add_track(".opus", |catalog, name| { + let mut config = AudioConfig::new(AudioCodec::Opus, 48_000, 2); + config.container = Container::Legacy; + catalog.lock().audio.renditions.insert(name, config); + }); + + // 50 ms video frames with a keyframe every second one, so the keyframes land + // on 0 and 100 ms exactly. + for i in 0..6u64 { + live.track.write(video_frame(i * 50_000, i % 2 == 0)).unwrap(); + } + // 20 ms Opus packets, so one lands on 100 ms exactly. + for i in 0..12u64 { + audio + .write(raw_frame(i * 20_000, &[0x08, 0xaa, 0xbb, 0xcc], true)) + .unwrap(); + } + live.track.finish().unwrap(); + audio.finish().unwrap(); + + let mut exporter = crate::container::fmp4::Export::new(live.source(), live.catalog_stream().await); + let init = fragment_now(&mut exporter).await; + assert!(init.init); + let (video_id, audio_id) = track_ids(&init.data); + + let fragments = drain_now(&mut exporter).await; + let fragments: Vec<(u32, usize)> = fragments + .iter() + .map(|fragment| { + let trafs = traf_samples(&fragment.data); + assert_eq!(trafs.len(), 1, "expected one traf per fragment"); + trafs[0] + }) + .collect(); + + // The first audio run is 0, 20, 40, 60 and 80 ms. The packet at 100 ms opens + // the second, beside the keyframe it shares an instant with; six here would + // mean it had been swept into the run before. + let first_audio = fragments + .iter() + .find(|(id, _)| *id == audio_id) + .expect("an audio fragment"); + assert_eq!( + first_audio.1, 5, + "the packet on the boundary belongs to the GOP it opens: {fragments:?}", + ); + let _ = video_id; +} + /// A video track that ends while the audio plays on must not leave its last GOP /// behind the whole rest of the audio. ///