diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 1037c315d0..f9cdc07452 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -107,9 +107,11 @@ the clock the catalog started with. A clock set with real start. Group starts never go backwards. A group starting before the previous group's -start ends the import with `TimestampRewind`, as a restarted encoder or a looping -file wrapping to the top does, flagged MPEG-TS discontinuity or not; republish -it as a new broadcast. Frames may still dip below the previous group's content: +start ends the import with `TimestampRewind`, whose `timestamp` and `floor` fields +name the refused frame and the previous group's start; the message gives both in +microseconds. A restarted encoder or a looping file wrapping to the top does this, +flagged MPEG-TS discontinuity or not; republish it as a new broadcast. Frames may +still dip below the previous group's content: B-frames, and a keyframe overlapping the previous group's last frame. A flagged MPEG-TS discontinuity that jumps forward continues the broadcast. diff --git a/quest/m1/README.md b/quest/m1/README.md index 1c7594d277..32fb9a4a8e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -60,7 +60,6 @@ QUIC studies there on that rule. - [Nightly tests on macOS and Windows](/quest/m1/nightly-platform-tests.md) - `moq-auth`, `hang`, `moq-tokio`, and `moq-native` tests run nightly where `Instant` and sockets differ, not just compile - [CI runner stalls](/quest/m1/ci-runner-stalls.md) - the 0.4 to 0.8 s freezes of both interop tracks on CI are attributed from a week of nightlies and fixed or told apart from playback bugs - [Catalog estimate rate](/quest/m1/catalog-estimate-rate.md) - a rising `jitter`/`delay` estimate republishes the catalog at most once a second, in js/publish and moq-mux -- [Rewind detail](/quest/m1/timestamp-rewind-detail.md) - a refused timestamp rewind names the frame's timestamp and the live edge - [Subscriber max-delay](/quest/m1/subscriber-max-delay.md) - subscriber staleness becomes `max_delay`; publisher retention stays `max_age` - [Wire compatibility](/quest/m1/wire-compat.md) - a nightly run tests this checkout against the last published release for tokens, session wire, and catalog/container - [Same-hop importers](/quest/m1/hop-aligned-import.md) - importers fed one stream publish identical groups and timestamps, so failover between a redundant pair survives diff --git a/quest/m1/timestamp-rewind-detail.md b/quest/m1/timestamp-rewind-detail.md deleted file mode 100644 index 5e0da051df..0000000000 --- a/quest/m1/timestamp-rewind-detail.md +++ /dev/null @@ -1,29 +0,0 @@ -# [XS] A timestamp rewind names the frame and the edge - -## Goal - -When `container::Producer::write` refuses a frame below the live edge, the -error says which timestamp it refused and where the edge was, so a report like -#4635 diagnoses itself from the error text alone. - -## Plan - -`container::TimestampRewind` (`rs/moq-mux/src/container/mod.rs`) is a unit -struct, raised at `rs/moq-mux/src/container/producer.rs` when -`frame.timestamp < live_edge`, and also constructed on the consumer side -(`rs/moq-mux/src/container/consumer.rs`); give those the same detail. Give it `timestamp` and `edge` fields and put -both, in microseconds, in its message, for example "frame timestamp 1790802494898431 -µs is below the live edge 1790802494898432 µs". moq-ffi surfaces the message -unchanged. - -#4635 could not be reproduced: FLAC and AAC share this path, and strictly -increasing timestamps with a cut per frame pass. A cut per frame makes every -frame a group start, so it refuses any backward step, while a dip inside an -open group is tolerated. Test: a FLAC `import::Track` with `cut(None)` per -frame and one frame 1 µs back reports both values. - -Public API: a unit struct gains fields, so breaking. Wire: none. - -## Related - -- [#4635](https://github.com/moq-dev/moq/issues/4635) - the report this makes diagnosable; it stays open until the reporter's sequence is known diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index f88bc263ae..2ec7eebb99 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -234,8 +234,10 @@ impl Consumer { Poll::Ready(Ok(Some(Event::Frame(frame)))) => { let seq = group.group.sequence; let ts = frame.timestamp; - if self.floor.is_some_and(|floor| ts.as_micros() < floor.as_micros()) { - return Poll::Ready(Err(TimestampRewind.into())); + if let Some(floor) = self.floor + && ts.as_micros() < floor.as_micros() + { + return Poll::Ready(Err(TimestampRewind { timestamp: ts, floor }.into())); } if self.start.is_none_or(|(start, _)| start != seq) { self.start = Some((seq, ts)); @@ -474,7 +476,7 @@ impl Consumer { // A later group with a media timestamp below the latest delivered group's start is malformed: // group starts never go backwards. Markers have no media timestamp, so they are not this check. fn poll_malformed(&mut self, waiter: &kio::Waiter) -> Result<(), F::Error> { - let Some((prev_group, edge)) = self.start else { + let Some((prev_group, floor)) = self.start else { return Ok(()); }; @@ -483,9 +485,9 @@ impl Consumer { continue; } if let Poll::Ready(Ok(min)) = group.poll_min_timestamp(waiter, &self.format) - && min.as_micros() < edge.as_micros() + && min.as_micros() < floor.as_micros() { - return Err(TimestampRewind.into()); + return Err(TimestampRewind { timestamp: min, floor }.into()); } } @@ -1174,7 +1176,15 @@ mod tests { write_group(&mut track, 1, &[ts(0)]); track.finish().unwrap(); let err = consumer.read().await.unwrap_err(); - assert!(matches!(err, crate::Error::TimestampRewind(_))); + assert!(matches!( + err, + crate::Error::TimestampRewind(TimestampRewind { timestamp, floor }) + if timestamp == ts(0) && floor == ts(100_000) + )); + assert_eq!( + err.to_string(), + "frame timestamp 0 µs is below the previous group's start 100000 µs" + ); } #[tokio::test] @@ -1315,7 +1325,15 @@ mod tests { track.finish().unwrap(); assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(200_000)); let err = consumer.read().await.unwrap_err(); - assert!(matches!(err, crate::Error::TimestampRewind(_))); + assert!(matches!( + err, + crate::Error::TimestampRewind(TimestampRewind { timestamp, floor }) + if timestamp == ts(50_000) && floor == ts(100_000) + )); + assert_eq!( + err.to_string(), + "frame timestamp 50000 µs is below the previous group's start 100000 µs" + ); } // ---- Empty payloads ---- diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index af7fed0436..ace0733329 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -123,8 +123,17 @@ pub struct InvalidEnd; /// A group starts before the previous group did: the source restarted, which is a new broadcast. #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] -#[error("frame timestamp is below the previous group's start")] -pub struct TimestampRewind; +#[error( + "frame timestamp {} µs is below the previous group's start {} µs", + .timestamp.as_micros(), + .floor.as_micros() +)] +pub struct TimestampRewind { + /// The refused frame's timestamp. + pub timestamp: moq_net::Timestamp, + /// The lowest allowed timestamp: the previous group's start. + pub floor: moq_net::Timestamp, +} /// Encode and decode media frames over a moq-lite group. /// diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index 15cccd0878..dd039d150f 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -403,17 +403,23 @@ where /// /// Group starts never go backwards: a keyframe below the start of the previous group, or any /// frame below the start of the group before its own, returns - /// [`TimestampRewind`](super::TimestampRewind) without writing, the way an oversized frame is - /// refused. That is a restart, which is a new broadcast. Frames may still dip below the - /// previous group's content: B-frames, open-GOP leading pictures, and a keyframe that - /// overlaps the previous group's last frame. + /// [`TimestampRewind`](super::TimestampRewind) with the refused timestamp and that start, + /// without writing, the way an oversized frame is refused. That is a restart, which is a new + /// broadcast. Frames may still dip below the previous group's content: B-frames, open-GOP + /// leading pictures, and a keyframe that overlaps the previous group's last frame. pub fn write(&mut self, frame: Frame) -> crate::Result<()> { let floor = match frame.keyframe { true => self.start, false => self.floor, }; - if floor.is_some_and(|floor| timestamp_lt(frame.timestamp, floor)) { - return Err(super::TimestampRewind.into()); + if let Some(floor) = floor + && timestamp_lt(frame.timestamp, floor) + { + return Err(super::TimestampRewind { + timestamp: frame.timestamp, + floor, + } + .into()); } // A keyframe cuts the previous group, using its timestamp as the boundary where the diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index 4430b6d179..a245680ca6 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -866,6 +866,43 @@ mod tests { import.finish().unwrap(); } + #[tokio::test(start_paused = true)] + async fn flac_cut_per_frame_rewind_names_timestamp_and_edge() { + let (broadcast, catalog) = new_broadcast(); + let config = crate::codec::flac::Config { + min_block_size: 4608, + max_block_size: 4608, + min_frame_size: 0, + max_frame_size: 0, + sample_rate: 48_000, + channel_count: 2, + bits_per_sample: 24, + total_samples: 0, + md5: [0; 16], + }; + let mut import = Track::audio( + broadcast.reserve_track("audio").unwrap(), + catalog.reserve(), + AudioInit::new(AudioFormat::Flac, config.description()), + ) + .unwrap(); + + let floor = 1_790_802_494_898_432; + for timestamp in [floor - 10_000, floor] { + import + .decode(b"flac frame", Some(Timestamp::from_micros(timestamp).unwrap())) + .unwrap(); + import.cut(None).unwrap(); + } + let err = import + .decode(b"flac frame", Some(Timestamp::from_micros(floor - 1).unwrap())) + .unwrap_err(); + assert!(matches!(err, crate::Error::TimestampRewind(_)), "{err:?}"); + let message = err.to_string(); + assert!(message.contains("1790802494898431 µs"), "{message}"); + assert!(message.contains("1790802494898432 µs"), "{message}"); + } + #[tokio::test(start_paused = true)] async fn aac_import_attaches_audio_specific_config() { let (broadcast, catalog) = new_broadcast();