fix(net): wake parked reads when stamped successors abort

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

quest/m1/README.md

@@ -51,7 +51,6 @@ QUIC studies there on that rule.
 - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport
 - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls
 - [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
 - [IETF FETCH abandonment](/quest/m1/ietf-fetch-abandonment.md) - the Rust moq-transport subscriber cancels a group FETCH once its last reader leaves

rs/moq-net/benches/track.rs

@@ -320,13 +320,54 @@ 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,
 	bench_parallel_write,
 	bench_aborted_scan,
 	bench_subscriber_churn,
 	bench_subscriber_churn_after_peak,
-	bench_subscriber_join
+	bench_subscriber_join,
+	bench_parked_read
 );
 criterion_main!(benches);

rs/moq-net/src/model/resume.rs

@@ -3721,6 +3721,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");

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

@@ -2589,10 +2589,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
@@ -3363,6 +3369,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 {
@@ -6009,6 +6020,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