diff --git a/quest/m1/README.md b/quest/m1/README.md index 69f008727b..07b4d10715 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -30,7 +30,6 @@ QUIC studies there on that rule. - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost - [Delete removed cluster flags](/quest/m1/cluster-shims.md) - once a release has carried their refusals, `mesh` and `linger` leave `cluster::Config` - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` -- [moq-mux truncated group](/quest/m1/mux-truncated-group.md) - a test pins that moq-mux skips a truncated spliced group as aborted - [Demand everywhere](/quest/m1/demand-everywhere.md) - requests and producers at every level watch subscribers through `demand()` alone, ahead of the bindings exposing it - [Merge queue](/quest/m1/merge-queue.md) - the required checks run on `merge_group`, so a stale green check can no longer break main - [Binding audio delay](/quest/m1/binding-surface.md) - moq-ffi and every wrapper configure and observe audio playout delay diff --git a/quest/m1/mux-truncated-group.md b/quest/m1/mux-truncated-group.md deleted file mode 100644 index fb44cad34f..0000000000 --- a/quest/m1/mux-truncated-group.md +++ /dev/null @@ -1,23 +0,0 @@ -# [XS] moq-mux skips a truncated spliced group - -## Goal - -A moq-mux test feeds its consumer a spliced group whose continuation can never -arrive and asserts the group is skipped as aborted, not read as a short clean -group. - -## Plan - -Start after #4689 lands: it makes `group::Consumer` reads and `finished()` fail -on a truncated spliced group (`Error::Dropped` for a pruned seam, else the dead -route's error), where they used to end clean. moq-mux's consumer already drops a group -whose read fails, but only code reading verified that; no test covers a spliced -group. Build one with the moq-net resume test fixtures, or the smallest public -path to a pruned seam. - -Public API: none. Wire: none. - -## Related - -- [#4689](https://github.com/moq-dev/moq/pull/4689) - the truncated-group behavior this test pins -- [#4491](https://github.com/moq-dev/moq/pull/4491) - the `resume.rs` splice path diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 2ec7eebb99..7f38cc050e 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -1643,6 +1643,91 @@ mod tests { assert!(frames.len() >= 4, "Expected >= 4 frames, got {}", frames.len()); } + #[tokio::test(start_paused = true)] + async fn truncated_spliced_group_skips_to_the_next_clean_group() { + let origin = crate::source::produce_origin(); + let hops = moq_net::Hops::try_from(vec![moq_net::Hop::new(10).unwrap()]).unwrap(); + let first_route = origin + .dynamic( + "live", + moq_net::origin::Route::default().with_hops(hops.clone()).with_cost(5), + ) + .unwrap(); + let pending = origin.consume().request_broadcast("live"); + let first = moq_net::broadcast::Info::new().produce(); + let info = hang::container::track_info(hang::catalog::PRIORITY.video); + let first_track = first.create_track("video", info.clone()).unwrap(); + first_route.requested_broadcast().await.unwrap().accept(&first); + let broadcast = pending.await.unwrap(); + let track = broadcast + .track("video") + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_age(Duration::from_secs(2))) + .await + .unwrap(); + let format = Container::Legacy(crate::container::Kind::Data); + let mut consumer = Consumer::new(track, Container::Legacy(crate::container::Kind::Data)); + + let mut open = first_track.create_group(0u64.into()).unwrap(); + format + .write( + &mut open, + &[Frame { + timestamp: ts(0), + payload: Bytes::from_static(&[0xDE, 0xAD]), + keyframe: true, + duration: None, + }], + ) + .unwrap(); + + // Each cheaper route advances beyond the seam without serving its continuation. + // Enough takeovers prune every route that could still cover group 0. + let mut routes = Vec::new(); + let mut sources = Vec::new(); + for sequence in 1..=4 { + let route = origin + .dynamic( + "live", + moq_net::origin::Route::default() + .with_hops(hops.clone()) + .with_cost(5 - sequence), + ) + .unwrap(); + let source = moq_net::broadcast::Info::new().produce(); + let mut track = source.create_track("video", info.clone()).unwrap(); + route.requested_broadcast().await.unwrap().accept(&source); + track.demand().used().await.unwrap(); + if sequence == 1 { + assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(0)); + } + write_group(&mut track, sequence, &[ts(sequence * 100_000)]); + routes.push(route); + sources.push((source, track)); + } + + let waiter = kio::Waiter::noop(); + + // Group 1 was never handed out before its segment was pruned. An abort + // skips straight to the retained successor; a short clean group emits GroupEnd. + let event = consumer.poll_event(&waiter); + let Poll::Ready(Ok(Some(Event::Frame(frame)))) = event else { + panic!( + "expected a successor frame, current={}; pending={}", + consumer.current, + event.is_pending() + ); + }; + assert_eq!(frame.timestamp, ts(200_000)); + assert!(frame.keyframe); + assert_eq!(consumer.current, 2); + assert!( + matches!(consumer.poll_event(&waiter), Poll::Ready(Ok(Some(Event::GroupEnd)))), + "the complete successor still emits its clean boundary" + ); + assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(300_000)); + } + // ---- Eviction recovery (pause/resume) ---- /// A group that aged out of the relay cache (aborted with `Error::Old`) while the