From cdbc167e479fa875ad25ab00e2f9d076857554c6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:29:59 -0700 Subject: [PATCH 1/2] chore: claim parked read wakes quest Co-Authored-By: GPT-6 From ab385775f5d7cc7895758fc465faaf6ff90a7c70 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 22:40:25 -0700 Subject: [PATCH 2/2] fix(net): wake parked reads when stamped successors abort Co-Authored-By: GPT-6 --- ...ic-tracks-and-preserve-sequences-across.md | 1 - quest/m1/README.md | 1 - quest/m1/parked-read-wakes.md | 30 ------- rs/moq-net/benches/track.rs | 43 +++++++++- rs/moq-net/src/model/resume.rs | 81 +++++++++++++++++++ rs/moq-net/src/model/track.rs | 50 +++++++++++- 6 files changed, 170 insertions(+), 36 deletions(-) delete mode 100644 quest/m1/parked-read-wakes.md diff --git a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md index 21eff93ccd..7ab04e037c 100644 --- a/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md +++ b/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md @@ -90,4 +90,3 @@ the closed producer's cache or terminal state. ## Related - [#4491](https://github.com/moq-dev/moq/pull/4491) - edits `resume.rs` and its takeover tests -- [Parked reads wake](/quest/m1/parked-read-wakes.md) - edits the same `resume.rs` wakeups diff --git a/quest/m1/README.md b/quest/m1/README.md index a29b80e341..8927e57e6c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -47,7 +47,6 @@ QUIC studies there on that rule. - [Browser close code](/quest/m1/browser-close-code.md) - a playwright case proves the page reads a relay's close code and reason, on every web-transport backend - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does -- [Parked reads wake](/quest/m1/parked-read-wakes.md) - a read parked on an evicted or aborted group wakes and re-judges, for resume successors and plain tracks alike - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls diff --git a/quest/m1/parked-read-wakes.md b/quest/m1/parked-read-wakes.md deleted file mode 100644 index 77b75479a2..0000000000 --- a/quest/m1/parked-read-wakes.md +++ /dev/null @@ -1,30 +0,0 @@ -# [S] Parked reads wake - -## Goal - -A read parked on a group that is evicted or aborted always wakes and -re-judges, for resume successors and plain tracks alike, pinned by mock-time -regression tests. - -## Plan - -Follow-ups from [#4484](https://github.com/moq-dev/moq/pull/4484), which made -`resume::Successor` recompute its answer on every judgment: - -- Its regression tests cover an aborted unstamped successor only. Eviction - wakes through a different path, so add a resume-level test where the - successor is evicted before its first frame. If it already passes, this - item lands as a test. -- A stamped group that is aborted without being evicted wakes no parked - reader, on a plain track as well as a successor. It only matters under a - timestamp rewind. Reproduce it first, then wake the waiters at the source. - -Settled: one quest, since both are the same wakeup invariant in -`rs/moq-net/src/model/track.rs` and `resume.rs`. - -Public API: none. Wire: none. - -## Related - -- [#4491](https://github.com/moq-dev/moq/pull/4491) - a resumed group holds demand on its replacement copy in the same `resume.rs` wakeups -- [#2991](/quest/m1/2991-net-coalesce-dynamic-tracks-and-preserve-sequences-across.md) - extends the `resume.rs` takeover tests diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 14f7661d3b..637157a266 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -320,6 +320,46 @@ fn bench_subscriber_join(c: &mut Criterion) { group.finish(); } +/// Parked reads re-judge their reach, including the successor's abort waiter. +/// Sweep readers and cached groups so added per-reader scans show up as a slope. +fn bench_parked_read(c: &mut Criterion) { + let mut group = c.benchmark_group("track_parked_read"); + for cached in [8, 64, 512] { + for readers in [1, 8, 64] { + let broadcast = broadcast::Info::default().produce(); + let track = broadcast.create_track("bench", None).unwrap(); + let mut head = track.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + for _ in 1..cached { + let mut next = track.append_group().unwrap(); + next.write_frame(Timestamp::ZERO, Bytes::from_static(b"next")).unwrap(); + next.finish().unwrap(); + } + let waiters: Vec<_> = (0..readers).map(|_| kio::Waiter::noop()).collect(); + let mut held: Vec<_> = waiters + .iter() + .map(|waiter| { + let mut sub = track.subscribe(track::Subscription::default().with_max_age(Duration::from_secs(1))); + let Poll::Ready(Ok(Some(mut head))) = sub.poll_recv_group(waiter) else { + panic!("head is cached"); + }; + assert!(matches!(head.poll_read_frame(waiter), Poll::Ready(Ok(Some(_))))); + head + }) + .collect(); + group.throughput(Throughput::Elements(readers as u64)); + group.bench_function(BenchmarkId::new(format!("cached_{cached}"), readers), |b| { + b.iter(|| { + for (head, waiter) in held.iter_mut().zip(&waiters) { + assert!(black_box(head.poll_read_frame(waiter)).is_pending()); + } + }); + }); + } + } + group.finish(); +} + criterion_group!( benches, bench_fanout, @@ -327,6 +367,7 @@ criterion_group!( bench_aborted_scan, bench_subscriber_churn, bench_subscriber_churn_after_peak, - bench_subscriber_join + bench_subscriber_join, + bench_parked_read ); criterion_main!(benches); diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 8e2672d9f5..ed86af6587 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -3626,6 +3626,87 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let (mut b, b_read) = track_pair("b"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let mut successor = b.create_group(1u64.into()).unwrap(); + successor + .write_frame(Duration::from_secs(30).try_into().unwrap(), b"b1".to_vec()) + .unwrap(); + write_group_at(&mut b, 2, "b2", Duration::from_secs(1)); + write_group_at(&mut b, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b_read, Position::group(1)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + successor.abort(Error::Cancel).unwrap(); + assert!( + counter.count() > before, + "the stamped successor'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:?}"); + } + + /// Cache eviction removes an unstamped successor from its track, so the next + /// segment's first group becomes the bound for the read already parked on A. + #[tokio::test] + async fn evicted_unstamped_successor_re_resolves_across_segments() { + tokio::time::pause(); + let (a, a_read) = track_pair("a"); + let pool = crate::cache::Pool::new(crate::cache::Config::default().with_expiry(Duration::from_secs(1))); + let info = broadcast::Info { + pool: pool.clone(), + ..Default::default() + }; + let b = track::Producer::new(Arc::new(info), "b", None); + let (mut c, c_read) = track_pair("c"); + let mut head = a.create_group(0u64.into()).unwrap(); + head.write_frame(Timestamp::ZERO, b"a0".to_vec()).unwrap(); + let successor = b.create_group(1u64.into()).unwrap(); + // The live edge is protected from eviction and lies beyond B's segment. + let _protected = b.create_group(4u64.into()).unwrap(); + write_group_at(&mut c, 2, "c2", Duration::from_secs(1)); + write_group_at(&mut c, 3, "edge", Duration::from_secs(20)); + let mut producer = Producer::new(); + producer.switch(a_read, None).unwrap(); + producer.switch(b.consume(), Position::group(1)).unwrap(); + producer.switch(c_read, Position::group(2)).unwrap(); + let mut sub = producer.consume().subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the unstamped successor leaves reach unbounded"); + assert_eq!(read(&mut reading), b"a0"); + + let (counter, waker) = CountWaker::new(); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + let before = counter.count(); + crate::model::clock::advance(Duration::from_secs(2)); + pool.gc(crate::model::clock::now()); + assert!(successor.is_aborted(), "cache GC evicted the successor"); + assert!( + b.consume().peek_group(1).is_none(), + "the slot was removed from the cache" + ); + assert!(counter.count() > before, "the successor's eviction lost its wakeup"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + #[tokio::test] async fn pruned_segment_boundary_is_judged_against_later_segments() { let (mut a, a_read) = track_pair("a"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index d526934dff..c4d6d4d9eb 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2626,10 +2626,16 @@ impl Consumer { ConsumerKind::Plain(state) => { let mut start = None; let _ = state.poll(waiter, |state| { - start = state.first_servable(from, cap).map(|group| { + while let Some(group) = state.first_servable(from, cap) { + // A stamped group can abort without changing its track. Register + // before checking it again, so an abort either wakes or re-resolves. + let _ = group.poll_closed(waiter); let _ = group.poll_timestamp(waiter); - group.timestamp() - }); + if !group.is_aborted() { + start = Some(group.timestamp()); + break; + } + } Poll::<()>::Pending }); start @@ -3397,6 +3403,11 @@ impl group::Expiry for GroupExpiry { let _ = self.state.poll(waiter, |state| { let budget = clamp_max_age(max_age, state.max_age_bound()); 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); + } let edge = state.drift_edge(cap, outer, successor); expired = state.is_stale(self.sequence, &edge, budget); if expired { @@ -6029,6 +6040,39 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } + /// A rewound successor can keep a drained group within budget until its abort + /// moves the reach to the next cached group, without changing the track itself. + #[tokio::test] + async fn aborted_stamped_successor_wakes_a_parked_read() { + tokio::time::pause(); + let mut producer = track_producer("test", None); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); + let mut successor = producer.append_group().unwrap(); + successor + .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice()) + .unwrap(); + append_at(&mut producer, 1000); + append_at(&mut producer, 20_000); + let mut sub = producer.subscribe(None); + let mut reading = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); + assert!(reading.read_frame().await.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!(reading.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + successor.abort(Error::Cancel).unwrap(); + assert!( + woken.load(Ordering::SeqCst), + "the stamped successor'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 first timestamp on a *newer* group can convict a held one, so the held reader has /// to be woken by it. The conviction needs a group beyond the held one's successor: /// a group is bounded by where its successor begins, so the successor itself never