From c29b7b74c2611357ff2a3ae6dfdc87e0e29169c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:05:15 -0700 Subject: [PATCH 1/3] fix(net): a group withheld for its first frame holds the end Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 40 +++++++++++++++++++++-- rs/moq-net/tests/track_tail.rs | 60 ++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c6c0716d64..b3e6598a73 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1348,12 +1348,19 @@ impl TrackState { // `sealed` also ends a locally closed receive track without a declared end. // An abort still wins unless that end had already settled: a group below it was still open. let reached = self.sealed - || self - .final_sequence - .is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin); + || self.final_sequence.is_some_and(|fin| { + self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin && !self.withholds(fin) + }); reached && (self.abort.is_none() || self.settled) } + /// Whether a group below `fin` is still withheld from readers until its first frame + /// lands (see [`Producer::receive_group`]): the end is not reached before it shows, or + /// readers would end without it. + fn withholds(&self, fin: u64) -> bool { + self.lookup.range(..fin).any(|(_, slot)| slot.pending) + } + /// Whether the declared end is reached and every cached group below it finished, /// so nothing the end promised is still in flight. fn is_settled(&self) -> bool { @@ -6448,6 +6455,33 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// A route can declare the end, then land every group's header before its first frame. + /// Those groups are withheld until their frames land, so the end is not reached before + /// they show: a reader ending at the boundary would never see them. + #[test] + fn a_withheld_group_holds_the_end() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(60))); + producer.finish_at(2).unwrap(); + let mut received: Vec<_> = (0..2) + .map(|sequence| producer.receive_group(group::Info { sequence }).unwrap()) + .collect(); + assert!( + subscriber.recv_group().now_or_never().is_none(), + "the end waits for the withheld groups" + ); + + for group in &mut received { + group + .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + producer.reveal_group(group); + } + assert_eq!(subscriber.assert_group().sequence, 0); + assert_eq!(subscriber.assert_group().sequence, 1); + assert!(subscriber.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs index b938b0430d..0eb66859f8 100644 --- a/rs/moq-net/tests/track_tail.rs +++ b/rs/moq-net/tests/track_tail.rs @@ -358,3 +358,63 @@ async fn a_lost_datagram_never_delays_the_end() { drop((pair, _keep)); } } + +/// A publisher can declare the end, then open each group's stream before writing its first +/// frame. The subscriber withholds a group until that frame lands, and the end waits for +/// it: a reader that ended at the boundary would miss every group. +#[moq_net_sim::test] +async fn a_group_whose_first_frame_trails_its_header_is_delivered() { + for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"] { + let Pair { + pair, + mut track, + remote, + _keep, + } = connect(version).await; + + let reader = moq_net_sim::spawn(async move { + let subscription = moq_net::track::Subscription::default() + .with_start(moq_net::track::Position::group(0)) + .with_max_delay(TIMEOUT); + let mut sub = remote + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let mut groups = Vec::new(); + while let Some(mut group) = sub.recv_group().await.expect("track aborted") { + let frame = group.read_frame().await.expect("group aborted").expect("a frame"); + assert_eq!(&frame.payload[..], PAYLOAD); + groups.push(group.sequence); + } + groups + }); + moq_net_sim::timeout(TIMEOUT, support::harness::subscribed(&track)) + .await + .expect("no subscriber appeared"); + + track.finish_at(2).unwrap(); + let mut groups = [track.append_group().unwrap(), track.append_group().unwrap()]; + // Paused time advances only once every task is idle: both headers and the end + // have reached the subscriber, ahead of any frame. + moq_net_sim::sleep(GRACE / 10).await; + assert!( + !reader.is_finished(), + "{version}: the track ended before its groups showed" + ); + + for group in &mut groups { + group.write_frame(Timestamp::ZERO, PAYLOAD).unwrap(); + group.finish().unwrap(); + } + let mut groups = moq_net_sim::timeout(TIMEOUT, reader) + .await + .expect("the subscription never ended") + .expect("reader panicked"); + // Arrival order: either group's frame may land first. + groups.sort(); + assert_eq!(groups, [0, 1], "{version}"); + drop((track, pair, _keep)); + } +} From f2bd8f3bb63888557787c3208d763e3bac526010 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:38:47 -0700 Subject: [PATCH 2/3] test(net): a withheld group that ends empty releases the end Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 36 +++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b3e6598a73..c68fa37a78 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -6482,6 +6482,42 @@ mod test { assert!(subscriber.recv_group().now_or_never().unwrap().unwrap().is_none()); } + /// A withheld group whose stream ends with no frame, cleanly or reset, shows when it + /// ends, which wakes a reader parked on the end. + #[moq_net_sim::test] + async fn a_withheld_group_that_ends_empty_releases_the_end() { + for abort in [false, true] { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(60))); + producer.finish_at(1).unwrap(); + let group = producer.receive_group(group::Info { sequence: 0 }).unwrap(); + let handle = group.clone(); + + let first = { + let mut next = std::pin::pin!(subscriber.recv_group()); + assert!(futures::poll!(next.as_mut()).is_pending(), "the end waits for the group"); + match abort { + true => group.abort(Error::Cancel).unwrap(), + false => group.finish().unwrap(), + } + producer.reveal_group(&handle); + moq_net_sim::timeout(Duration::from_secs(1), next) + .await + .expect("the reveal wakes the reader") + .unwrap() + }; + + match abort { + // A reset group shows nothing, so the reader ends. + true => assert!(first.is_none()), + false => { + assert_eq!(first.expect("the empty group").sequence, 0); + assert!(subscriber.recv_group().now_or_never().unwrap().unwrap().is_none()); + } + } + } + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is From bd6876faca9bbbb2c039087f9f2a3086e6773ce5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:39:08 -0700 Subject: [PATCH 3/3] style: rustfmt Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c68fa37a78..23db600959 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -6495,7 +6495,10 @@ mod test { let first = { let mut next = std::pin::pin!(subscriber.recv_group()); - assert!(futures::poll!(next.as_mut()).is_pending(), "the end waits for the group"); + assert!( + futures::poll!(next.as_mut()).is_pending(), + "the end waits for the group" + ); match abort { true => group.abort(Error::Cancel).unwrap(), false => group.finish().unwrap(),