diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index eacd270e3c..5517ce98fb 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -3304,9 +3304,15 @@ impl group::Expiry for GroupExpiry { loop { // An abort can change reach even after the successor is stamped. // Register before judging, so a racing abort is observed or wakes us. - if let Some(group) = state.first_servable(self.sequence.saturating_add(1), cap) { - let _ = group.poll_closed(waiter); - } + // An abort only closes the group, never touching the track, so one that + // landed before registering must re-select or the replacement goes unwatched. + let successor = loop { + let successor = state.first_servable(self.sequence.saturating_add(1), cap); + match successor { + Some(group) if group.poll_closed(waiter).is_ready() && group.is_aborted() => continue, + successor => break successor, + } + }; let edge = state.drift_edge(cap); expired = state.is_stale(self.sequence, &edge, budget); if expired { @@ -3316,8 +3322,11 @@ impl group::Expiry for GroupExpiry { // A first timestamp can change the verdict without mutating the track: // on a group past the edge (a new edge), or on the candidate's // unstamped immediate successor (a reach where there was none). - // Register on the candidate and every unstamped servable group above - // it. If one raced this scan, resolve the edge again before Pending. + // Register on the candidate, its successor, and every servable group + // past the edge. If one raced this scan, resolve the edge again before + // Pending. Groups between the successor and the edge can move neither, + // and walking them would cost each of N parked serves the whole + // backlog, O(N^2) per track change. let mut timestamp_raced = false; if let Some(slot) = state.lookup.get(&self.sequence) { let group = &slot.group; @@ -3328,17 +3337,18 @@ impl group::Expiry for GroupExpiry { timestamp_raced = true; } } - for (_, slot) in state + let past = edge + .presentation + .map_or(self.sequence, |live| live.sequence.max(self.sequence)); + let beyond = state .lookup - .range((std::ops::Bound::Excluded(self.sequence), std::ops::Bound::Unbounded)) - { - let group = &slot.group; - if !super::subscription::before_end(group.sequence, cap) { - break; - } - if slot.visible - && !group.is_aborted() - && group.timestamp().is_none() + .range((std::ops::Bound::Excluded(past), std::ops::Bound::Unbounded)) + .map(|(_, slot)| slot) + .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) + .filter(|slot| slot.visible && !slot.group.is_aborted()) + .map(|slot| &slot.group); + for group in successor.into_iter().chain(beyond) { + if group.timestamp().is_none() && group.poll_timestamp(waiter).is_ready() && group.timestamp().is_some() { @@ -5827,6 +5837,104 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } + /// A parked read registers only where a first timestamp can change its verdict: its + /// immediate successor and the groups past the edge. Walking every group between them + /// costs each parked publisher stream the whole backlog, enough for a 2s audio backlog + /// at 400 groups/s to pin a publisher's runtime and starve its connection. + #[test] + fn a_parked_read_ignores_first_frames_between_its_successor_and_the_edge() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10))); + 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 + let mut between = producer.append_group().unwrap(); + append_at(&mut producer, 3000); // the edge + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0); + 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()); + + between + .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!( + !woken.load(Ordering::SeqCst), + "a first frame below the edge moves neither the reach nor the edge" + ); + + // Control: a first frame past the edge is still observed. + let mut beyond = producer.append_group().unwrap(); + assert!(next.as_mut().poll(&mut cx).is_pending()); + woken.store(false, Ordering::SeqCst); + beyond + .write_frame(Timestamp::from_millis(4000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!(woken.load(Ordering::SeqCst), "a new edge wakes the parked read"); + } + + /// An aborted successor hands the reach to the next group, which sits below the edge + /// and so is watched only because the read re-selects its successor. + #[test] + fn a_parked_read_watches_the_replacement_for_an_aborted_successor() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + let successor = producer.append_group().unwrap(); + let mut replacement = producer.append_group().unwrap(); + append_at(&mut producer, 20_000); // the edge + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0, "an unstamped successor leaves the reach unbounded"); + 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()); + + successor.abort(Error::Cancel).unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the successor's abort wakes the parked read" + ); + woken.store(false, Ordering::SeqCst); + assert!( + next.as_mut().poll(&mut cx).is_pending(), + "the unstamped replacement leaves the reach unbounded" + ); + + replacement + .write_frame(Timestamp::from_millis(1000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the replacement's first frame bounds the reach" + ); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// 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