fix(net): a parked read watches its edge abort

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

@@ -1024,12 +1024,13 @@ impl Producer {
 		}
 	}
 
-	/// Block until the group is closed or aborted.
+	/// Block until the group is aborted, including by eviction. Finishing does not close it.
 	pub async fn closed(&self) -> Error {
 		kio::wait(|waiter| self.poll_closed(waiter)).await
 	}
 
-	/// Poll until the group is closed or aborted; ready with the cause.
+	/// Poll until the group is aborted, including by eviction; ready with the cause.
+	/// Finishing does not close it.
 	pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<Error> {
 		self.state.poll_closed(waiter).map(|()| self.abort_reason())
 	}
@@ -1484,9 +1485,10 @@ impl Consumer {
 		self.keep_alive();
 	}
 
-	/// Park `waiter` until the group closes (finish, abort, or eviction). Subscribers
-	/// register on parked groups so an eviction wakes them; a group that already closed
-	/// cleanly can never abort, so no waiter is needed.
+	/// Park `waiter` until the group closes: an abort (including eviction), or its last
+	/// producer dropping. Finishing does not close it, since a finished group can still be
+	/// evicted. Subscribers register on parked groups so an eviction wakes them; a group
+	/// that closed without an abort can never abort, so no waiter is needed.
 	pub(crate) fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<()> {
 		self.cursor.state.poll_closed(waiter)
 	}

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

@@ -809,8 +809,8 @@ impl TrackState {
 	}
 
 	/// [`Self::drifted`] under `cap`, parking `waiter` on exactly what else can move the
-	/// verdict: the successor's first frame or abort, and the edge crossing the deadline.
-	/// Parking on every track change instead wakes each of N parked reads per append.
+	/// verdict: the successor's first frame or abort, the edge's abort, and the edge
+	/// crossing the deadline. Parking on every track change instead wakes each of N parked reads per append.
 	///
 	/// The caller parks on a group landing above `sequence` first, in the way that fits
 	/// whether this track holds it.
@@ -841,7 +841,21 @@ impl TrackState {
 		// Registered before the edge is resolved, so a write crossing the deadline is
 		// either seen here or wakes us.
 		self.cache.wakes().watch_deadline(reach, budget, waiter);
-		self.drifted(sequence, &self.drift_edge(cap), budget)
+		loop {
+			let edge = self.drift_edge(cap);
+			// The edge's abort presents nothing and never touches the track, yet hands the
+			// edge to a lower group that may already sit past the deadline. Watch it, and
+			// re-resolve if it already landed. Terminates: an aborted group is never the
+			// edge again, so each pass resolves a lower one.
+			if let Some(live) = edge.presentation
+				&& let Some(slot) = self.lookup.get(&live.sequence)
+				&& slot.group.poll_closed(waiter).is_ready()
+				&& slot.group.is_aborted()
+			{
+				continue;
+			}
+			return self.drifted(sequence, &edge, budget);
+		}
 	}
 
 	/// Resolve a one-shot fetch from the track side: the cached group, or an [`Error`]
@@ -6458,6 +6472,44 @@ mod test {
 		assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
 	}
 
+	/// A rewound edge's abort hands the edge back to a lower group that already presented
+	/// past the deadline. No frame write crosses it afterwards, and the abort never touches
+	/// the track, so only the read watching the edge's abort sees the head go stale.
+	#[test]
+	fn a_parked_read_watches_its_rewound_edge_abort() {
+		let mut producer = track_producer("test", None);
+		let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(5)));
+		let mut head = producer.append_group().unwrap();
+		head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
+			.unwrap();
+		append_at(&mut producer, 1000); // the successor bounds the head's reach
+		append_at(&mut producer, 10_000); // past the deadline, but below the edge
+		let mut edge = producer.append_group().unwrap();
+		edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x"))
+			.unwrap();
+		edge.finish().unwrap();
+
+		let mut held = subscriber
+			.recv_group()
+			.now_or_never()
+			.unwrap()
+			.unwrap()
+			.expect("head group");
+		assert_eq!(held.sequence, 0, "the rewound edge keeps the head within budget");
+		assert!(held.read_frame().now_or_never().unwrap().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!(held.read_frame());
+		assert!(next.as_mut().poll(&mut cx).is_pending());
+
+		edge.abort(Error::Cancel).unwrap();
+		assert!(woken.load(Ordering::SeqCst), "the edge'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 received group's frames present while it is hidden, so a parked read they wake
 	/// still sees the older edge and parks again. Revealing the group has to wake it once
 	/// more, though the read sits below the group's nearest shown predecessor.