diff --git a/quest/m1/README.md b/quest/m1/README.md index 484124788a8..b86d06f9229 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -187,7 +187,6 @@ per-frame CPU growth since #5926. - [moqsink reports track demand only once the track is in the catalog](/quest/m1/moqsink-demand-catalog.md) - A `moqsink` request pad's `track-demand` stays null until its track is listed in the published catalog, and only then reports `wanted` - [Ordered reads can skip owed holes, and snapshot latest() does](/quest/m1/ordered-owed-holes.md) - moq-net's `Ordered` gains a read that skips past an owed hole to the newest buffered group, while keeping the reader's bounds - [A Rust relay never replays from group 0 to a lite-03/04 subscriber](/quest/m1/relay-inflight-replay.md) - A Rust relay or publisher serving a lite-03 or lite-04 SUBSCRIBE from a track copy that is still resolving starts at the newest group once the copy resolves -- [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 - [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 ## Related 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 8ba675d9c37..ae543920738 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 035e0c8e4b8..8e941a3b681 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() { @@ -2084,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}; @@ -2155,6 +2154,72 @@ 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"); + 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 + .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();