From 45bc74252c712eb3a12a71d63b7afed2031e41ae Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 14 Oct 2026 10:47:22 -0700 Subject: [PATCH 1/2] chore: claim export source reads quest From ad4ec5c3f716b0d43dc3fbe2051448dfcc3119f1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 14 Oct 2026 11:55:42 -0700 Subject: [PATCH 2/2] fix(mux): consume export gaps and ends in read order Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 - quest/m1/export-source-read.md | 32 ---- rs/moq-mux/src/container/consumer.rs | 201 ++++++++++---------------- rs/moq-mux/src/container/source.rs | 176 +++++++++++++++++++--- rs/moq-mux/src/container/ts/import.rs | 12 +- 5 files changed, 240 insertions(+), 182 deletions(-) delete mode 100644 quest/m1/export-source-read.md diff --git a/quest/m1/README.md b/quest/m1/README.md index ba4384d9868..af3ed6b3ff3 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -143,7 +143,6 @@ above them. - [JS request deadline](/quest/m1/js-request-deadline.md) - each JS request, PUBLISH_NAMESPACE included, has one fatal 10 s timer from create to answer and fails fast without stream credit, so nothing queues or retries - [Handshake deadline covers admission](/quest/m1/handshake-admission-deadline.md) - `listen.timeout` bounds a connection through auth admission on both listeners, replacing the hardcoded admission timeout - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust -- [ExportSource reads Read items](/quest/m1/export-source-read.md) - exporters learn of gaps and ends from the read stream, and the private counters go - [Rust request credit](/quest/m1/rs-request-credit.md) - a moq-net request fails at once without stream credit instead of waiting - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome - [#933](/quest/m1/933-video-rotation-metadata-not-propagated-from-mobile-camera.md) - the catalog rotation follows the live camera's orientation diff --git a/quest/m1/export-source-read.md b/quest/m1/export-source-read.md deleted file mode 100644 index 0195d0ad9b4..00000000000 --- a/quest/m1/export-source-read.md +++ /dev/null @@ -1,32 +0,0 @@ -# [XS] ExportSource reads gaps and ends as Read items - -## Goal - -`moq_mux::container::ExportSource` (`rs/moq-mux/src/container/source.rs`) -consumes the container read's `Read::{Frame, Gap, End}` items instead of -polling the consumer's crate-private `discontinuity()` and `markers()` -counters, and those counters are deleted. - -## Plan - -#5288 made gaps and declared ends items in the read stream but left -`ExportSource` on the old counters, so the exporters (fMP4, TS, MKV, FLV, and -the raw H.264/H.265 exports) still learn of a gap after the fact. - -- Map `Gap` to the exporter's discontinuity handling and `End` to its - declared-end handling, in delivery order. A join reads no `Gap` (#5311), - though `discontinuity()` still counts losses before the first item; a fresh - exporter has nothing to cut there. -- Keep the events `Read` filters reaching the exporters. fMP4 cuts a fragment - on `Event::GroupEnd` while the track stays live, and sizes a durationless - final sample from a trailing `Event::FrameEnd { opens: false }`; losing - either stalls fMP4 or writes the wrong duration. Keep a crate-private path - for both alongside the Read items. Tests: a live fMP4 export still emits a - fragment at each group end, and - `zero_fragment_duration_waits_for_the_exact_video_endpoint` passes. -- Delete `Consumer::discontinuity()` and `Consumer::markers()` once nothing - reads them, along with `ExportSource::markers()` if it only forwards. -- Existing exporter tests must pass unchanged; add one where a gap and an end - arrive back to back to pin the order. - -Public API: none (crate-private). Wire: none. diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index fd17dc84983..f909e3ca088 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -17,6 +17,13 @@ pub(crate) enum Event { GroupEnd, } +/// Read items and exporter-only boundaries in delivery order. +pub(crate) enum Item { + Read(Read), + FrameEnd(Timestamp), + GroupEnd, +} + /// Decode a moq-lite track into a stream of media [`Frame`]s in age-bounded /// presentation order. /// @@ -98,12 +105,6 @@ pub struct Consumer { // contiguity check on a delivered hole. presented_end: Option, - // Increments on a declared marker, an unproven delivered hole, and a latency skip. - discontinuity: u64, - - // Increments on a declared marker only: the publisher paused or broke its timeline forward. - markers: u64, - // Media went missing since the last item read, so the next one is preceded by a gap. gap: bool, @@ -170,28 +171,12 @@ impl Consumer { start: None, floor: None, presented_end: None, - discontinuity: 0, - markers: 0, gap: false, next: None, group: 0, } } - /// A counter that increments at each playhead event: a declared marker group, an - /// unproven delivered hole, a latency skip, or an evicted current group. - pub(crate) fn discontinuity(&self) -> u64 { - self.discontinuity - } - - /// A counter that increments only when the publisher declared a marker group: a pause or - /// a forward break in its timeline, since a rewind is refused. A hole or a latency skip - /// moves the playhead ([`Self::discontinuity`]) but the timeline carries on, so this - /// stays put. - pub(crate) fn markers(&self) -> u64 { - self.markers - } - /// The group sequence of the last event read. pub(crate) fn group(&self) -> u64 { self.group @@ -226,37 +211,45 @@ impl Consumer { /// Uses a single waiter that gets registered on all relevant kio channels, /// avoiding the need for `tokio::select!` or `FuturesUnordered`. pub fn poll_read(&mut self, waiter: &kio::Waiter) -> Poll, F::Error>> { - if let Some(read) = self.next.take() { - return Poll::Ready(Ok(Some(read))); - } loop { - // Where delivered media reached before this item, which moves the edge. - let edge = self.edge; - let read = match ready!(self.poll_event(waiter))? { - Some(Event::Frame(frame)) => Read::Frame { - group: self.group, - frame, - }, - Some(Event::FrameEnd { end, opens: true }) => Read::End { group: self.group, end }, - // A trailing endpoint is the last frame's duration, which only the skip logic needs. - Some(Event::FrameEnd { opens: false, .. } | Event::GroupEnd) => continue, + match ready!(self.poll_item(waiter))? { + Some(Item::Read(read)) => return Poll::Ready(Ok(Some(read))), + Some(Item::FrameEnd(_) | Item::GroupEnd) => continue, None => return Poll::Ready(Ok(None)), - }; - let resume = match &read { - Read::Frame { frame, .. } => frame.timestamp, - Read::End { end, .. } => *end, - Read::Gap { .. } => unreachable!("events never read as a gap"), - }; - if std::mem::take(&mut self.gap) - && let Some(edge) = edge - { - self.next = Some(read); - return Poll::Ready(Ok(Some(gap(edge, resume)))); } - return Poll::Ready(Ok(Some(read))); } } + /// Read items while preserving boundaries needed to finish exporter fragments. + pub(crate) fn poll_item(&mut self, waiter: &kio::Waiter) -> Poll, F::Error>> { + if let Some(read) = self.next.take() { + return Poll::Ready(Ok(Some(Item::Read(read)))); + } + let edge = self.edge; + let read = match ready!(self.poll_event(waiter))? { + Some(Event::Frame(frame)) => Read::Frame { + group: self.group, + frame, + }, + Some(Event::FrameEnd { end, opens: true }) => Read::End { group: self.group, end }, + Some(Event::FrameEnd { end, opens: false }) => return Poll::Ready(Ok(Some(Item::FrameEnd(end)))), + Some(Event::GroupEnd) => return Poll::Ready(Ok(Some(Item::GroupEnd))), + None => return Poll::Ready(Ok(None)), + }; + let resume = match &read { + Read::Frame { frame, .. } => frame.timestamp, + Read::End { end, .. } => *end, + Read::Gap { .. } => unreachable!("events never read as a gap"), + }; + if std::mem::take(&mut self.gap) + && let Some(edge) = edge + { + self.next = Some(read); + return Poll::Ready(Ok(Some(Item::Read(gap(edge, resume))))); + } + Poll::Ready(Ok(Some(Item::Read(read)))) + } + /// Read media or a clean group boundary without waiting for a successor group. pub(crate) fn poll_event(&mut self, waiter: &kio::Waiter) -> Poll, F::Error>> { // Grab any new groups from the track, recording whether the track is finished. @@ -361,17 +354,12 @@ impl Consumer { } // Cleanly finished group: advance to the next sequence. Poll::Ready(Ok(None)) => { - let marker = group.marker(); if let Some(end) = group.max_end { self.presented_end = Some(end); } self.pending.pop_front(); self.current += 1; self.note_group_edge(); - if marker { - self.bump_playhead(); - self.markers += 1; - } return Poll::Ready(Ok(Some(Event::GroupEnd))); } } @@ -466,7 +454,6 @@ impl Consumer { if hole || had_marker || discarded { self.lose(); } - self.markers += u64::from(had_marker); let new_current = self.pending.front().map(|g| g.sequence).unwrap(); tracing::debug!(old = self.current, new = new_current, "skipping slow groups"); @@ -503,13 +490,8 @@ impl Consumer { } } - fn bump_playhead(&mut self) { - self.discontinuity += 1; - } - // Media this reader will never deliver: the next item is preceded by a gap. fn lose(&mut self) { - self.bump_playhead(); self.gap = true; } @@ -912,6 +894,21 @@ mod tests { Ok(frames) } + /// Read a continuous sequence, refusing gaps and declared stops. + async fn continuous_frames(consumer: &mut Consumer) -> Vec + where + F::Error: std::fmt::Debug, + { + let mut frames = Vec::new(); + while let Some(read) = consumer.read().await.unwrap() { + let Read::Frame { frame, .. } = read else { + panic!("unexpected break: {read:?}") + }; + frames.push(frame); + } + frames + } + /// Wrap `track` so only the container consumer performs age skips. The /// transport gets the media track's full retention window and hands over every /// retained group; what the container does with them is what the test measures. @@ -1132,7 +1129,7 @@ mod tests { } #[tokio::test] - async fn empty_group_declares_a_discontinuity() { + async fn a_marker_group_declares_an_end() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let consumer_track = track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(2))); @@ -1143,16 +1140,12 @@ mod tests { write_group(&mut track, 2, &[ts(1_000_000)]); track.finish().unwrap(); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(0)); - assert_eq!(consumer.discontinuity(), 0); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000_000)); - assert_eq!(consumer.discontinuity(), 1); + assert_eq!(items(&mut consumer).await, ["0", "end 0", "1000000"]); } - /// A missing group moves the playhead, but the timeline carries on: it is not a marker. - /// A declared marker is both. + /// Missing media reads as a gap; only the publisher can declare an end. #[tokio::test] - async fn only_a_marker_breaks_the_timeline() { + async fn only_a_marker_declares_an_end() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let consumer_track = track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(2))); @@ -1165,13 +1158,10 @@ mod tests { write_group(&mut track, 4, &[ts(2_000_000)]); track.finish().unwrap(); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(0)); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000_000)); - assert_eq!(consumer.discontinuity(), 1, "the hole moves the playhead"); - assert_eq!(consumer.markers(), 0, "but the timeline carried on"); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(2_000_000)); - assert_eq!(consumer.discontinuity(), 2); - assert_eq!(consumer.markers(), 1, "the marker declares a break"); + assert_eq!( + items(&mut consumer).await, + ["0", "gap 0+1000000", "1000000", "end 1000000", "2000000"] + ); } #[tokio::test] @@ -1190,13 +1180,11 @@ mod tests { write_group(&mut track, 2, &[ts(1_000)]); track.finish().unwrap(); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(0)); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000)); - assert_eq!(consumer.discontinuity(), 0, "an empty group is not a marker"); + assert_eq!(items(&mut consumer).await, ["0", "1000"]); } #[tokio::test] - async fn latency_skip_preserves_empty_group_discontinuity() { + async fn latency_skip_reports_a_gap() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let consumer_track = track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(2))); @@ -1208,14 +1196,7 @@ mod tests { write_group(&mut track, 3, &[ts(1_000_000)]); track.finish().unwrap(); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(0)); - assert_eq!(consumer.discontinuity(), 0); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000_000)); - assert_eq!( - consumer.discontinuity(), - 1, - "a shed marker is a timestamp hole, so the playhead jumps" - ); + assert_eq!(items(&mut consumer).await, ["0", "gap 0+1000000", "1000000"]); } #[tokio::test] @@ -1463,12 +1444,11 @@ mod tests { let mut consumer = Consumer::new(track.subscribe(None), Container::Legacy(crate::container::Kind::Data)); write_group(&mut track, 0, &[ts(0), ts(66_000), ts(33_000)]); track.finish().unwrap(); - let frames = read_all(&mut consumer).await.unwrap(); + let frames = continuous_frames(&mut consumer).await; assert_eq!( frames.iter().map(|f| f.timestamp).collect::>(), vec![ts(0), ts(66_000), ts(33_000)] ); - assert_eq!(consumer.discontinuity(), 0); } #[tokio::test] @@ -1510,16 +1490,15 @@ mod tests { write_group(&mut track, 0, &[ts(0), ts(33_000)]); write_group(&mut track, 1, &[ts(66_000), ts(50_000)]); track.finish().unwrap(); - let frames = read_all(&mut consumer).await.unwrap(); + let frames = continuous_frames(&mut consumer).await; assert_eq!( frames.iter().map(|f| f.timestamp.as_micros()).collect::>(), vec![0, 33_000, 66_000, 50_000] ); - assert_eq!(consumer.discontinuity(), 0); } #[tokio::test] - async fn a_marker_group_and_forward_jump_bumps_playhead_once() { + async fn a_marker_group_and_forward_jump_reads_one_end() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let mut consumer = Consumer::new( track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(2))), @@ -1529,14 +1508,11 @@ mod tests { write_marker_group(&mut track, 1, ts(0)); write_group(&mut track, 2, &[ts(1_000_000)]); track.finish().unwrap(); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(0)); - assert_eq!(consumer.discontinuity(), 0); - assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000_000)); - assert_eq!(consumer.discontinuity(), 1); + assert_eq!(items(&mut consumer).await, ["0", "end 0", "1000000"]); } #[tokio::test] - async fn a_zero_budget_idle_resume_jumps_the_playhead_after_a_shed_marker() { + async fn a_zero_budget_idle_resume_reports_the_shed_marker_as_a_gap() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let mut consumer = Consumer::new(track.subscribe(None), Container::Legacy(crate::container::Kind::Audio)); consumer.max_delay = Duration::ZERO; @@ -1545,16 +1521,12 @@ mod tests { write_marker_group(&mut track, 1, ts(0)); write_group(&mut track, 2, &[ts(1_000_000)]); track.finish().unwrap(); + assert!(matches!(consumer.read().await.unwrap(), Some(Read::Gap { .. }))); assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(1_000_000)); - assert_eq!( - consumer.discontinuity(), - 1, - "a shed marker with a timestamp jump is a playhead event" - ); } #[tokio::test] - async fn a_zero_budget_skip_keeps_a_contiguous_marker() { + async fn a_zero_budget_skip_reports_a_contiguous_shed_marker_as_a_gap() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let mut consumer = Consumer::new(track.subscribe(None), Container::Legacy(crate::container::Kind::Audio)); consumer.max_delay = Duration::ZERO; @@ -1576,12 +1548,8 @@ mod tests { write_marker_group(&mut track, 1, ts(10_000)); write_group(&mut track, 2, &[ts(10_000)]); + assert!(matches!(consumer.read().await.unwrap(), Some(Read::Gap { .. }))); assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(10_000)); - assert_eq!( - consumer.discontinuity(), - 1, - "a drained marker still declares the encoder's break" - ); group0.finish().unwrap(); } @@ -3018,27 +2986,20 @@ mod tests { track.finish().unwrap(); - let frames = tokio::time::timeout(Duration::from_secs(2), async { - let mut frames = Vec::new(); - while let Some(frame) = consumer.read_frame().await.unwrap() { - frames.push(frame); - } - frames - }) - .await - .expect("consumer hung — duration skip regression"); + let frames = tokio::time::timeout(Duration::from_secs(2), async { continuous_frames(&mut consumer).await }) + .await + .expect("consumer hung — duration skip regression"); assert_eq!(frames.len(), 2); assert_eq!(frames[0].timestamp, ts(0)); assert_eq!(frames[1].timestamp, ts(33_000)); - assert_eq!(consumer.discontinuity(), 0, "a contiguous duration skip is not a hole"); // group0 is intentionally never finished. drop(group0); } #[tokio::test] - async fn a_nonsequential_contiguous_jump_does_not_bump_playhead() { + async fn a_nonsequential_contiguous_jump_has_no_gap() { let track = track_producer("test", timed()); let mut consumer = Consumer::new( track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))), @@ -3055,12 +3016,8 @@ mod tests { write_duration_frame(&mut group1, ts(33_000), ts(33_000)); group1.finish().unwrap(); track.finish().unwrap(); - let mut frames = Vec::new(); - while let Some(frame) = consumer.read_frame().await.unwrap() { - frames.push(frame); - } + let frames = continuous_frames(&mut consumer).await; assert_eq!(frames.len(), 2); - assert_eq!(consumer.discontinuity(), 0); } /// When the current group's frame ends before the next group begins, there is diff --git a/rs/moq-mux/src/container/source.rs b/rs/moq-mux/src/container/source.rs index 9035e56d57d..dccea4b3bfc 100644 --- a/rs/moq-mux/src/container/source.rs +++ b/rs/moq-mux/src/container/source.rs @@ -19,12 +19,12 @@ use std::task::{Poll, ready}; use bytes::Bytes; use hang::catalog::{AudioConfig, Container, Reference, VideoCodec, VideoConfig}; -use super::consumer::Event; +use super::consumer::{Event, Item}; use crate::catalog::hang::Container as HangContainer; use crate::codec::annexb::{InBand, InBandCodec}; use crate::codec::h264::Avc1; use crate::codec::h265::Hvc1; -use crate::container::{Consumer, Frame}; +use crate::container::{Consumer, Frame, Read}; /// Per-track video transform that bridges between codec shapes. pub(crate) enum VideoTransform { @@ -89,6 +89,10 @@ pub(crate) struct ExportSource { /// The transform absorbed a keyframe that carried no picture (parameter sets sent /// as their own frame), so the next frame it emits is the sync point. absorbed_keyframe: bool, + skips: u64, + markers: u64, + // A stopped encoder may still emit final packets in this group on the old timeline. + ends: Option<(u64, u64)>, } impl ExportSource { @@ -170,6 +174,9 @@ impl ExportSource { video_codec: Some(config.codec.clone()), video_dimensions: catalog_dimensions(config), absorbed_keyframe: false, + skips: 0, + markers: 0, + ends: None, }; source.resolve_video_dimensions(&[])?; Ok(Some(source)) @@ -202,6 +209,9 @@ impl ExportSource { video_codec: None, video_dimensions: None, absorbed_keyframe: false, + skips: 0, + markers: 0, + ends: None, })) } @@ -227,6 +237,9 @@ impl ExportSource { video_codec: None, video_dimensions: None, absorbed_keyframe: false, + skips: 0, + markers: 0, + ends: None, }) } @@ -254,13 +267,9 @@ impl ExportSource { } } - /// The underlying consumer's playhead generation, or 0 until the subscription resolves: - /// it counts skipped groups as well as markers ([`Consumer::discontinuity`]). + /// Gaps and declared ends this source has advanced past. pub fn skips(&self) -> u64 { - match &self.state { - SourceState::Active(consumer) => consumer.discontinuity(), - _ => 0, - } + self.skips } /// The resolved codec-config record, if available. @@ -286,18 +295,9 @@ impl ExportSource { } } - /// How many marker groups the publisher declared, each a pause or forward break in its - /// timeline, or 0 until the subscription resolves. - /// - /// See [`Consumer::markers`]. Sample it alongside each frame returned by - /// [`poll_read`](Self::poll_read): the frame read while the counter changes is the - /// first past the break, so anything anchored on the media clock has to re-anchor to - /// it. A skipped group is not a marker: the timeline carries on. + /// Declared ends this source has advanced past. pub fn markers(&self) -> u64 { - match &self.state { - SourceState::Active(consumer) => consumer.markers(), - _ => 0, - } + self.markers } /// True if the codec config is resolved (either present in the catalog, @@ -387,14 +387,39 @@ impl ExportSource { let SourceState::Active(consumer) = &mut self.state else { unreachable!("subscription resolved into an Active consumer"); }; - let event = ready!(consumer.poll_event(waiter))?; + let event = ready!(consumer.poll_item(waiter))?; if std::mem::take(&mut self.live) { // Started at the live edge: from here the budget tolerates reordering. consumer.set_max_delay(self.max_delay); } match event { - Some(Event::Frame(frame)) => frame, - event => return Poll::Ready(Ok(event)), + Some(Item::Read(Read::Frame { group, frame })) => { + if self.ends.is_some_and(|(end_group, _)| end_group != group) { + self.finish_ends(); + } + frame + } + Some(Item::Read(Read::Gap { .. })) => { + self.finish_ends(); + self.skips += 1; + return Poll::Ready(Ok(Some(Event::GroupEnd))); + } + Some(Item::Read(Read::End { group, end })) => { + if self.ends.is_some_and(|(end_group, _)| end_group != group) { + self.finish_ends(); + } + self.ends.get_or_insert((group, 0)).1 += 1; + return Poll::Ready(Ok(Some(Event::FrameEnd { end, opens: true }))); + } + Some(Item::FrameEnd(end)) => return Poll::Ready(Ok(Some(Event::FrameEnd { end, opens: false }))), + Some(Item::GroupEnd) => { + self.finish_ends(); + return Poll::Ready(Ok(Some(Event::GroupEnd))); + } + None => { + self.finish_ends(); + return Poll::Ready(Ok(None)); + } } }; @@ -428,6 +453,13 @@ impl ExportSource { } } + fn finish_ends(&mut self) { + if let Some((_, count)) = self.ends.take() { + self.skips += count; + self.markers += count; + } + } + fn refresh_description(&mut self) { // Track the transform's record even after it is first set: a mid-stream // reconfiguration rebuilds the avcC/hvcC with a new parameter set, and the @@ -622,4 +654,104 @@ mod tests { assert_eq!(observed, max_delay); } + async fn audio_export() -> (moq_net::track::Producer, ExportSource) { + use std::time::Duration; + let live = Live::audio(audio(None)); + let mut export = ExportSource::for_audio( + &live.source().await, + live.track.name(), + &audio(None), + Duration::from_secs(10), + ) + .unwrap() + .unwrap(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let consumer = Consumer::new( + track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))), + HangContainer::Legacy(crate::container::Kind::Audio), + ); + export.state = SourceState::Active(Box::new(consumer)); + (track, export) + } + + #[tokio::test(start_paused = true)] + async fn a_gap_then_end_reaches_the_exporter_in_order() { + use crate::container::Container as _; + use crate::container::test_util::raw_frame; + + let (track, mut export) = audio_export().await; + let wire = HangContainer::Legacy(crate::container::Kind::Audio); + for (sequence, timestamp, payload) in [(0, 0, b"audio".as_slice()), (2, 1_000_000, b"".as_slice())] { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + wire.write(&mut group, &[raw_frame(timestamp, payload, true)]).unwrap(); + group.finish().unwrap(); + } + track.finish().unwrap(); + let waiter = kio::Waiter::noop(); + assert!(matches!( + export.poll_event(&waiter), + Poll::Ready(Ok(Some(Event::Frame(_)))) + )); + assert_eq!((export.skips(), export.markers()), (0, 0)); + assert!(matches!( + export.poll_event(&waiter), + Poll::Ready(Ok(Some(Event::GroupEnd))) + )); + assert_eq!((export.skips(), export.markers()), (0, 0)); + assert!(matches!( + export.poll_event(&waiter), + Poll::Ready(Ok(Some(Event::GroupEnd))) + )); + assert_eq!((export.skips(), export.markers()), (1, 0), "the gap precedes the end"); + assert!(matches!( + export.poll_event(&waiter), + Poll::Ready(Ok(Some(Event::FrameEnd { opens: true, .. }))) + )); + assert_eq!((export.skips(), export.markers()), (1, 0)); + assert!(matches!( + export.poll_event(&waiter), + Poll::Ready(Ok(Some(Event::GroupEnd))) + )); + assert_eq!( + (export.skips(), export.markers()), + (2, 1), + "the end is crossed without a successor frame" + ); + } + #[tokio::test(start_paused = true)] + async fn final_audio_packets_keep_the_stopped_encoders_timeline() { + use crate::container::Container as _; + use crate::container::test_util::raw_frame; + + let (track, mut export) = audio_export().await; + let wire = HangContainer::Legacy(crate::container::Kind::Audio); + for (sequence, frames) in [ + (0, vec![raw_frame(0, b"audio", true)]), + ( + 1, + vec![raw_frame(50_000, b"", true), raw_frame(40_000, b"drain", false)], + ), + (2, vec![raw_frame(100_000, b"resumed", true)]), + ] { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + wire.write(&mut group, &frames).unwrap(); + group.finish().unwrap(); + } + track.finish().unwrap(); + let waiter = kio::Waiter::noop(); + for (timestamp, generation) in [(0, 0), (40_000, 0), (100_000, 1)] { + let Poll::Ready(Ok(Some(frame))) = export.poll_read(&waiter) else { + panic!("frame ready") + }; + assert_eq!(frame.timestamp.as_micros(), timestamp); + assert_eq!( + (export.skips(), export.markers()), + (generation, generation), + "only the next group starts a new timeline" + ); + } + } } diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 36a4cd7fc5f..daad1e569c6 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -6099,12 +6099,14 @@ pub(super) mod test { crate::catalog::hang::Container::Legacy(crate::container::Kind::Audio), ); let mut frames = Vec::new(); - while let Ok(Ok(Some(frame))) = - tokio::time::timeout(std::time::Duration::from_millis(50), reader.read_frame()).await - { - frames.push(frame); + let mut breaks = 0; + while let Ok(Ok(Some(read))) = tokio::time::timeout(std::time::Duration::from_millis(50), reader.read()).await { + match read { + crate::container::Read::Frame { frame, .. } => frames.push(frame), + crate::container::Read::Gap { .. } | crate::container::Read::End { .. } => breaks += 1, + } } - (frames, reader.discontinuity()) + (frames, breaks) } /// Both [`two_stream_import`] renditions' frames and break counts, in catalog order.