fix(net): wake every read a fill answer parks, so a drop poll can't strand it

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

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) {

rs/moq-net/src/model/track/miss.rs

@@ -97,15 +97,16 @@ pub(super) struct Misses {
 pub(super) type GivenUp = Vec<ops::Range<u64>>;
 
 /// 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<BTreeSet<u64>>,
-	task: Mutex<Option<Waker>>,
+	tasks: Mutex<kio::WaiterList>,
 }
 
 /// 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<u64> {
 		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>) {
+				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();