diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 1ea02ec174..40bfefbc77 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1024,12 +1024,13 @@ impl Producer { } } - /// Block until the group is closed or aborted. + /// Block until the group is aborted, including by eviction. Finishing does not close it. pub async fn closed(&self) -> Error { kio::wait(|waiter| self.poll_closed(waiter)).await } - /// Poll until the group is closed or aborted; ready with the cause. + /// Poll until the group is aborted, including by eviction; ready with the cause. + /// Finishing does not close it. pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll { self.state.poll_closed(waiter).map(|()| self.abort_reason()) } @@ -1484,9 +1485,10 @@ impl Consumer { self.keep_alive(); } - /// Park `waiter` until the group closes (finish, abort, or eviction). Subscribers - /// register on parked groups so an eviction wakes them; a group that already closed - /// cleanly can never abort, so no waiter is needed. + /// Park `waiter` until the group closes: an abort (including eviction), or its last + /// producer dropping. Finishing does not close it, since a finished group can still be + /// evicted. Subscribers register on parked groups so an eviction wakes them; a group + /// that closed without an abort can never abort, so no waiter is needed. pub(crate) fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<()> { self.cursor.state.poll_closed(waiter) } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 2f29e882dd..3f88398321 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -809,8 +809,8 @@ impl TrackState { } /// [`Self::drifted`] under `cap`, parking `waiter` on exactly what else can move the - /// verdict: the successor's first frame or abort, and the edge crossing the deadline. - /// Parking on every track change instead wakes each of N parked reads per append. + /// verdict: the successor's first frame or abort, the edge's abort, and the edge + /// crossing the deadline. Parking on every track change instead wakes each of N parked reads per append. /// /// The caller parks on a group landing above `sequence` first, in the way that fits /// whether this track holds it. @@ -841,7 +841,21 @@ impl TrackState { // Registered before the edge is resolved, so a write crossing the deadline is // either seen here or wakes us. self.cache.wakes().watch_deadline(reach, budget, waiter); - self.drifted(sequence, &self.drift_edge(cap), budget) + loop { + let edge = self.drift_edge(cap); + // The edge's abort presents nothing and never touches the track, yet hands the + // edge to a lower group that may already sit past the deadline. Watch it, and + // re-resolve if it already landed. Terminates: an aborted group is never the + // edge again, so each pass resolves a lower one. + if let Some(live) = edge.presentation + && let Some(slot) = self.lookup.get(&live.sequence) + && slot.group.poll_closed(waiter).is_ready() + && slot.group.is_aborted() + { + continue; + } + return self.drifted(sequence, &edge, budget); + } } /// Resolve a one-shot fetch from the track side: the cached group, or an [`Error`] @@ -6458,6 +6472,44 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// A rewound edge's abort hands the edge back to a lower group that already presented + /// past the deadline. No frame write crosses it afterwards, and the abort never touches + /// the track, so only the read watching the edge's abort sees the head go stale. + #[test] + fn a_parked_read_watches_its_rewound_edge_abort() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(5))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + append_at(&mut producer, 1000); // the successor bounds the head's reach + append_at(&mut producer, 10_000); // past the deadline, but below the edge + let mut edge = producer.append_group().unwrap(); + edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + edge.finish().unwrap(); + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0, "the rewound edge keeps the head within budget"); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + edge.abort(Error::Cancel).unwrap(); + assert!(woken.load(Ordering::SeqCst), "the edge's abort lost its wakeup"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// A received group's frames present while it is hidden, so a parked read they wake /// still sees the older edge and parks again. Revealing the group has to wake it once /// more, though the read sits below the group's nearest shown predecessor.