From 078922e498f3ca81a1db8df489a235308936b999 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:07:18 -0800 Subject: [PATCH 1/3] chore(quest): claim quest/m1/resume-model-7542.md Quest-Claim: 36143c3cd43ea1d0 From 5c88a5041df06ca30341df95b981d788a0c4d4c8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:34:16 -0800 Subject: [PATCH 2/3] fix(net): wake every read a fill answer parks, so a drop poll can't strand it #7542 gave a cursor's misses one waker for every fill it waits on, replaced by whichever path resolved last. A read parked behind its idle gate and a drop poll resolve with different waiters, so the drop poll's took over: the fill's answer woke only it, the drop poll settled the want, and the read's gate stayed armed on a want already gone. The read never reached the track's end. Register each resolving waiter in a kio::WaiterList instead, so an answer wakes every waiter that resolved since the last one, as polling each fill directly did before #7542. The resume model passes 3000 cases on seeds 1 to 16, 777, 4242, 12345 and its default again. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/resume-model-7542.md | 23 ------- rs/moq-net/src/model/resume_model.rs | 70 +++++++++++++++++++++ rs/moq-net/src/model/track/miss.rs | 92 +++++++++++++++++++++++----- 4 files changed, 146 insertions(+), 40 deletions(-) delete mode 100644 quest/m1/resume-model-7542.md diff --git a/quest/m1/README.md b/quest/m1/README.md index d8a307f3e3e..09fa90740cd 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -189,7 +189,6 @@ per-frame CPU growth since #5926. - [Poll-first async](/quest/m1/poll-first/README.md) - library crates compile no coroutines by default: async wrappers are generated from `poll_*` and gated behind an `async` feature - [Authored tracks default to a 30 s max-age](/quest/m1/max-age-defaults.md) - raw and media tracks advertise 30 s unless their author chooses otherwise, with explicit unlimited in every binding - [HLS export lists a same-codec reconfigure's first group under the new init](/quest/m1/hls-reinit-same-codec.md) - the hand-over pick, whichever of the catalog and the records arrives first, after the catalog `hls` removal is planned -- [The resume model passes its long run again](/quest/m1/resume-model-7542.md) - fix #7542's regression that drops groups a long-budget reader can still get - [A resume more than one group behind never hangs](/quest/m1/ietf-resume-behind.md) - a resumed read gets every group or a drop on every wire version - [moq-quic's idle connections keep no burst-sized state](/quest/m1/quic-readable-dedupe.md) - readable events are deduplicated and a quiet connection keeps no per-peak-stream bytes - [Stats name the route instance's epoch with a type](/quest/m1/stats-epoch-type.md) - `GroupFrame::epoch` gets a dedicated type and tests stop mislabelling the instance diff --git a/quest/m1/resume-model-7542.md b/quest/m1/resume-model-7542.md deleted file mode 100644 index 3eae4d31d94..00000000000 --- a/quest/m1/resume-model-7542.md +++ /dev/null @@ -1,23 +0,0 @@ -# [S] The resume model passes its long run again - -## Goal - -`moq-net`'s route-switching resume model -(`model::resume_model::readers_match_the_logical_track`) passes at 3000 cases -on every seed again, fixed at its cause, so a reader with a long budget never -gets groups back as dropped that its route can still deliver. - -## Plan - -Planned 2026-11-02 (maintainer, follow-up planning after the 2026-11-02 /quest-complete run), from [moq-dev/moq#7625](https://github.com/moq-dev/moq/pull/7625)'s -checks. - -- It fails on `main` on seeds 12345, 777 and 4242 with the same minimal cases, - and bisects to [moq-dev/moq#7542](https://github.com/moq-dev/moq/pull/7542) - (lite-07 frame bounds, timed datagram admission, sparse reverse search, and - shared time-anchor walks). The 64-case run inside `just check` still passes. -- Symptom: a reader with an hour budget gets groups 0 to 10 back as dropped - when the route only wrote group 11. -- Repro: `MOQ_RESUME_SEED=12345 MOQ_RESUME_CASES=3000 cargo test -p moq-net - --all-features --release --lib model::resume_model::readers_match_the_logical_track`. -- Keep #7542's flat per-range costs; add the minimal case as a fixed regression. diff --git a/rs/moq-net/src/model/resume_model.rs b/rs/moq-net/src/model/resume_model.rs index 1f2f2128155..2e8a71aeac8 100644 --- a/rs/moq-net/src/model/resume_model.rs +++ b/rs/moq-net/src/model/resume_model.rs @@ -1154,6 +1154,76 @@ fn regressions() { ], finish: false, }, + // A read parked on a fill wakes once it answers, though the drop poll resolved after + // it parked: the track's end with groups the fill gave up ends the read, so groups + // landing below the end afterwards are not owed. + Case { + ops: vec![ + Op::Up { instance: 1, slot: 0 }, + Op::Join { + reader: 0, + spec: spec(false, Budget::Hour, None, true), + }, + Op::Write { + instance: 1, + slot: 0, + group: Some(11), + frames: 1, + }, + Op::End { instance: 1, slot: 0 }, + ], + finish: false, + }, + // The same, once the serving route's fill of a group the reader already holds gives + // it up: the read ends with the track. + Case { + ops: vec![ + Op::Join { + reader: 0, + spec: spec(false, Budget::Hour, None, true), + }, + Op::Write { + instance: 0, + slot: 0, + group: Some(8), + frames: 1, + }, + Op::Up { instance: 0, slot: 1 }, + Op::GiveUp { + instance: 0, + slot: 1, + group: 8, + }, + ], + finish: true, + }, + Case { + ops: vec![ + Op::Write { + instance: 0, + slot: 0, + group: None, + frames: 1, + }, + Op::Join { + reader: 0, + spec: spec(false, Budget::Hour, None, true), + }, + Op::Up { instance: 0, slot: 1 }, + Op::Write { + instance: 0, + slot: 0, + group: None, + frames: 1, + }, + Op::GiveUp { + instance: 0, + slot: 1, + group: 1, + }, + ], + finish: true, + }, ]; for case in cases { if let Err(err) = run(&case) { diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 5fd4639dd4b..74474b80ac4 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -97,15 +97,16 @@ pub(super) struct Misses { pub(super) type GivenUp = Vec>; /// The joins whose fills changed their answer, recorded by each join's own waiter before it -/// wakes the reader's task. +/// wakes the reader's tasks. /// -/// The task is the one the reader last resolved for. A reader polled by another task since -/// still learns of every answer, since the joins' waiters stay registered and wake whichever -/// task the reader resolved for last. +/// Every waiter the reader resolved with since the last answer wakes, as each would had it +/// polled the fills itself: a read parked behind its idle gate and a drop poll each resolve +/// with a waiter of their own, and an answer only one of them saw would leave the other +/// parked on a want already settled. #[derive(Default)] struct Answered { keys: Mutex>, - task: Mutex>, + tasks: Mutex, } /// Wakes the reader once join `key`'s fill answers, recording which join it was. @@ -128,10 +129,8 @@ impl Wake for JoinWake { .unwrap_or_else(PoisonError::into_inner) .insert(self.key); // Woken with no lock held: an executor may poll inline. - let task = answered.task.lock().unwrap_or_else(PoisonError::into_inner).clone(); - if let Some(task) = task { - task.wake(); - } + let mut tasks = answered.tasks.lock().unwrap_or_else(PoisonError::into_inner).take(); + tasks.wake(); } } @@ -476,13 +475,12 @@ impl Misses { waiter: &kio::Waiter, ) -> RangeSet { let mut retry = RangeSet::new(); - { - let mut task = self.answered.task.lock().unwrap_or_else(PoisonError::into_inner); - if !task.as_ref().is_some_and(|task| task.will_wake(waiter.waker())) { - *task = Some(waiter.waker().clone()); - } - } - // Taken after the task is set: an answer landing since wakes the task set above. + self.answered + .tasks + .lock() + .unwrap_or_else(PoisonError::into_inner) + .register(waiter); + // Taken after registering: an answer landing since wakes the waiter registered above. let keys = std::mem::take(&mut *self.answered.keys.lock().unwrap_or_else(PoisonError::into_inner)); let mut keys = keys.into_iter().peekable(); while let Some(&key) = keys.peek() { @@ -2062,6 +2060,68 @@ mod tests { ); } + /// A read parked on a fill wakes once it answers, though a drop poll resolved with a + /// waiter of its own after the read parked: the read then sees the fill give up its + /// group and ends with the track. + #[kio::test] + async fn an_answered_fill_wakes_a_read_a_drop_poll_resolved_after() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[derive(Default)] + struct Count(AtomicUsize); + impl Wake for Count { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + let broadcast = crate::broadcast::Info::default().produce(); + let mut track = broadcast.create_track("fills", crate::track::Info::default()).unwrap(); + let requests = track.requests(); + track.start_at(8).unwrap(); + track + .create_group(crate::group::Info { sequence: 8 }) + .unwrap() + .finish() + .unwrap(); + track.finish().unwrap(); + // Group 0 lies below where the feed starts: only a fill may deliver it. + let ranges = [crate::track::Range::groups(0..1), crate::track::Range::groups(8..)]; + let mut reader = track.subscribe( + Subscription::default() + .with_delay(Delay::default().with_max(Duration::from_secs(60))) + .with_ranges(ranges), + ); + let (read, drops) = (Arc::new(Count::default()), Arc::new(Count::default())); + let read_waiter = kio::Waiter::new(Waker::from(read.clone())); + let drops_waiter = kio::Waiter::new(Waker::from(drops.clone())); + let mut handed = Vec::new(); + while let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&read_waiter) { + handed.push(group.sequence); + } + assert_eq!(handed, [8]); + assert!( + reader.poll_recv_group(&read_waiter).is_pending(), + "the read waits on the fill for group 0" + ); + assert!(reader.poll_drop(&drops_waiter).is_pending(), "nothing given up yet"); + + let fill = std::future::poll_fn(|cx| requests.poll_requested_fill(&kio::Waiter::new(cx.waker().clone()))) + .await + .unwrap(); + let (read_woken, drops_woken) = (read.0.load(Ordering::Relaxed), drops.0.load(Ordering::Relaxed)); + fill.drop_groups(fill.start().group..fill.end()); + assert!(read.0.load(Ordering::Relaxed) > read_woken, "the parked read wakes"); + assert!(drops.0.load(Ordering::Relaxed) > drops_woken, "the drop poll wakes"); + + // The drop poll takes the answer first, as a publisher forwarding drops may. + assert!(matches!(reader.poll_drop(&drops_waiter), Poll::Ready((groups, _)) if groups == (0..1))); + assert!( + matches!(reader.poll_recv_group(&read_waiter), Poll::Ready(Ok(None))), + "the read ends once the fill gave up the last group it owed" + ); + } + #[test] fn deferred_splits_and_replacements_retire_the_old_age_entries() { let mut deferred = Deferred::default(); From 2273648f8c94ac9d4850255420e3b634d2fc2a3f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 13:30:29 -0800 Subject: [PATCH 3/3] test(net): poll drops with a no-op waiter while a read waits on a fill A no-op poll must leave the real registrations in place, so the answer still wakes the parked read and the drop poll. Also note why the neighboring test wakes only the last task: the read resolves through its idle gate, which holds one task waker. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track/miss.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 6450352ccb4..8e941a3b681 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -2082,7 +2082,8 @@ mod tests { /// A reader waiting on many fills looks only at the ones that answered, yet each still /// wakes it, and wakes the task that read it last: a reader read by another task since - /// parked learns of the answer too. + /// parked learns of the answer too. The read resolves through its idle gate, which wakes + /// only the task that parked it last. #[kio::test] async fn an_answered_fill_wakes_the_task_that_read_last() { use std::sync::atomic::{AtomicUsize, Ordering}; @@ -2198,6 +2199,10 @@ mod tests { "the read waits on the fill for group 0" ); assert!(reader.poll_drop(&drops_waiter).is_pending(), "nothing given up yet"); + assert!( + reader.poll_drop(&kio::Waiter::noop()).is_pending(), + "a no-op poll leaves the real registrations in place" + ); let fill = std::future::poll_fn(|cx| requests.poll_requested_fill(&kio::Waiter::new(cx.waker().clone()))) .await