fix(net): keep a settled track's groups when it is aborted

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

@@ -411,15 +411,29 @@ impl TrackState {
 		next_sequence: u64,
 		end_sequence: Option<u64>,
 	) -> Poll<Result<Option<group::Producer>>> {
+		// An abort before the end settled cut the track off. One after it is a clean
+		// end (see `is_complete`): the cached groups were everything the end promised,
+		// so the cursor ends as `poll_recv_group` does, not with the abort.
+		if let Some(err) = &self.abort
+			&& !self.settled
+		{
+			return Poll::Ready(Err(err.clone()));
+		}
+		// Nothing more can arrive once the last producer is gone: the track was sealed
+		// (dropped with its end declared) or aborted with that end settled. Either closed
+		// the channel, so parking is not an option: the consumer would be handed the
+		// abort, or `Dropped`, instead of the end it reached.
+		let closed = self.sealed || self.abort.is_some();
+
 		// If the exclusive end is already at or below where we'd resume, no
 		// group can ever satisfy this call until the cap rises. Pending (not
 		// None) so the consumer is parked rather than told the stream is over.
 		// An empty range (`end == 0`) parks even at the first sequence.
 		if let Some(end) = end_sequence
 			&& end <= next_sequence
 		{
-			if let Some(err) = &self.abort {
-				return Poll::Ready(Err(err.clone()));
+			if closed {
+				return Poll::Ready(Ok(None));
 			}
 			return Poll::Pending;
 		}
@@ -440,16 +454,12 @@ impl TrackState {
 		}
 
 		// No in-range group is cached. Decide whether more could ever arrive.
-		if let Some(err) = &self.abort {
-			return Poll::Ready(Err(err.clone()));
-		}
 		// `final_sequence` is one past the last possible sequence. If our
 		// floor is already at/past it, nothing else can land in range.
-		// A sealed track produces nothing more either: the last producer
-		// dropped with the boundary already declared, so a gap below it is
-		// the end, not a wait. Cached in-range groups were returned above.
-		// This is the cursor the ordered and spliced readers use.
-		if self.sealed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) {
+		// A closed track produces nothing more either, so a gap below its
+		// boundary is the end, not a wait. Cached in-range groups were
+		// returned above. This is the cursor the ordered and spliced readers use.
+		if closed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) {
 			return Poll::Ready(Ok(None));
 		}
 		Poll::Pending
@@ -1195,8 +1205,14 @@ fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) {
 	// Decided before the cache goes: the groups below the end are the evidence.
 	state.settled = state.is_settled();
 	state.abort = Some(err);
-	state.clear_cache();
-	state.datagrams.clear();
+	// A settled end is a clean end: the track holds every group it promised, so keep
+	// them for consumers still draining, as `finish()` does. Clearing here would abort
+	// the finished groups a slower reader has not pulled yet, and it would then see the
+	// track end short of them with no error.
+	if !state.settled {
+		state.clear_cache();
+		state.datagrams.clear();
+	}
 	state.close();
 }
 
@@ -1506,6 +1522,11 @@ impl Producer {
 	/// of the leftover cache. Child groups are independent: a consumer that already pulled
 	/// a [`group::Consumer`] keeps its own handle and can finish reading it.
 	///
+	/// Unless the declared end had settled: the final sequence from
+	/// [`finish_at`](Self::finish_at) was reached and every group below it finished. The
+	/// track then already holds everything it promised, so it ends cleanly and keeps the
+	/// cached groups for consumers still draining, as [`finish`](Self::finish) does.
+	///
 	/// [`finish`](Self::finish) is deliberately not terminal: it declares the final
 	/// sequence, and lower-numbered groups may still be written afterwards.
 	pub fn abort(self, err: Error) -> Result<()> {
@@ -4116,8 +4137,9 @@ impl Ordered {
 	/// the cap rises or is removed.
 	///
 	/// Returns `Poll::Ready(Ok(Some(group)))` when a group is available,
-	/// `Poll::Ready(Ok(None))` when the track is finished,
-	/// `Poll::Ready(Err(e))` when the track has been aborted, or
+	/// `Poll::Ready(Ok(None))` when the track is finished (including an abort after its
+	/// declared end settled, see [`Producer::abort`]),
+	/// `Poll::Ready(Err(e))` when the track has been aborted short of its end, or
 	/// `Poll::Pending` when no group is available yet.
 	pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
 		self.inner.poll_next_group(waiter)
@@ -6953,6 +6975,83 @@ mod test {
 		assert!(matches!(res, Ok(None)));
 	}
 
+	/// The settled end stands on the ordered cursor too: a reader that starts after the
+	/// abort gets every group below the end, then the clean end, not the abort.
+	#[tokio::test]
+	async fn abort_after_the_end_settles_ends_clean_for_an_ordered_reader() {
+		let mut producer = track_producer("test", None);
+		let mut consumer = producer.subscribe(None).ordered();
+
+		for sequence in 0..2 {
+			let group = producer.create_group(group::Info { sequence }).unwrap();
+			group.finish().unwrap();
+		}
+		producer.finish_at(2).unwrap();
+		producer.abort(Error::Timeout).unwrap();
+
+		assert_eq!(drain_ordered(&mut consumer), [0, 1]);
+		let end = consumer
+			.next_group()
+			.now_or_never()
+			.expect("should not block")
+			.map(|group| group.map(|group| group.sequence));
+		assert!(matches!(end, Ok(None)), "expected the clean end, got {end:?}");
+	}
+
+	/// The settled end stands on the ordered cursor even where it has nothing to read:
+	/// capped below the end, and with a gap below the cap. Parking is not an option on a
+	/// closed track (the channel would turn it into the abort), so these end clean.
+	#[tokio::test]
+	async fn abort_after_the_end_settles_ends_clean_below_the_cap() {
+		let mut producer = track_producer("test", None);
+		let mut consumer = producer.subscribe(None).ordered();
+		for sequence in 0..2 {
+			producer
+				.create_group(group::Info { sequence })
+				.unwrap()
+				.finish()
+				.unwrap();
+		}
+		producer.finish_at(2).unwrap();
+		producer.abort(Error::Timeout).unwrap();
+
+		consumer.set_groups(..1);
+		assert_eq!(drain_ordered(&mut consumer), [0]);
+		let end = consumer
+			.next_group()
+			.now_or_never()
+			.expect("should not block")
+			.map(|group| group.map(|group| group.sequence));
+		assert!(
+			matches!(end, Ok(None)),
+			"capped at the end: expected the clean end, got {end:?}"
+		);
+
+		let mut producer = track_producer("test", None);
+		let mut consumer = producer.subscribe(None).ordered();
+		for sequence in [0, 2] {
+			producer
+				.create_group(group::Info { sequence })
+				.unwrap()
+				.finish()
+				.unwrap();
+		}
+		producer.finish_at(3).unwrap();
+		producer.abort(Error::Timeout).unwrap();
+
+		consumer.set_groups(..2);
+		assert_eq!(drain_ordered(&mut consumer), [0]);
+		let end = consumer
+			.next_group()
+			.now_or_never()
+			.expect("should not block")
+			.map(|group| group.map(|group| group.sequence));
+		assert!(
+			matches!(end, Ok(None)),
+			"gap below the cap: expected the clean end, got {end:?}"
+		);
+	}
+
 	#[tokio::test]
 	async fn recv_group_finishes_without_waiting_for_gaps() {
 		let producer = track_producer("test", None);

rs/moq-net/tests/settled_end_survives_abort.rs

@@ -0,0 +1,89 @@
+//! A track whose declared end has settled holds every group it promised: the end was
+//! reached and each group below it finished. An abort that lands afterwards, such as
+//! the session dying, ends the track cleanly but must not throw those groups away: a
+//! consumer that has not read them yet still gets them, then the clean end.
+
+mod support;
+
+use moq_net::{Error, Hop, Timestamp};
+
+fn produce_origin(hop: u64) -> moq_net::origin::Producer {
+	let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap()));
+	tokio::spawn(support::harness::run(driver));
+	producer
+}
+
+const GROUPS: u64 = 2;
+
+/// Publish `GROUPS` finished groups, declare the end at `GROUPS`, then abort the track.
+/// Returns the sequences a consumer that starts reading only now receives, in arrival
+/// or in sequence order, and how the read ended.
+async fn round(abort: bool, ordered: bool) -> (Vec<u64>, Option<Error>) {
+	let origin = produce_origin(1);
+	let broadcast = origin.create_broadcast("bcast").unwrap();
+	let mut track = broadcast.create_track("video", None).unwrap();
+	// From the first group with a replay window: a late reader is owed the whole track,
+	// not the live edge.
+	let subscription = moq_net::track::Subscription::default()
+		.with_start(moq_net::track::Position::group(0))
+		.with_max_age(std::time::Duration::from_secs(30));
+	let mut consumer = track.subscribe(subscription);
+
+	for _ in 0..GROUPS {
+		let mut group = track.append_group().unwrap();
+		group.write_frame(Timestamp::ZERO, b"frame".as_slice()).unwrap();
+		group.finish().unwrap();
+	}
+	track.finish_at(GROUPS).unwrap();
+	// Every group below the end is finished: the end has settled and the track is
+	// complete. Nothing has been read yet.
+	if abort {
+		track.abort(Error::Session(moq_net::SessionError::Cancel)).unwrap();
+	}
+
+	let mut got = Vec::new();
+	if ordered {
+		let mut consumer = consumer.ordered();
+		loop {
+			match consumer.next_group().await {
+				Ok(Some(group)) => got.push(group.sequence),
+				Ok(None) => return (got, None),
+				Err(err) => return (got, Some(err)),
+			}
+		}
+	}
+	loop {
+		match consumer.recv_group().await {
+			Ok(Some(group)) => got.push(group.sequence),
+			Ok(None) => return (got, None),
+			Err(err) => return (got, Some(err)),
+		}
+	}
+}
+
+fn assert_whole((got, err): (Vec<u64>, Option<Error>), what: &str) {
+	assert!(
+		err.is_none() && got == (0..GROUPS).collect::<Vec<_>>(),
+		"{what}: got {got:?} of {GROUPS} groups, err={err:?} (every group had finished before \
+		 the end, so a late reader gets all of them and then the clean end)"
+	);
+}
+
+/// The groups a settled track holds outlive an abort: a late reader gets all of them.
+#[tokio::test]
+async fn an_abort_after_the_end_settled_keeps_the_groups_for_a_late_reader() {
+	assert_whole(round(true, false).await, "arrival order, aborted after the end settled");
+}
+
+/// The same in sequence order: the ordered cursor ends clean too, not with the abort.
+#[tokio::test]
+async fn an_abort_after_the_end_settled_keeps_the_groups_for_a_late_ordered_reader() {
+	assert_whole(round(true, true).await, "sequence order, aborted after the end settled");
+}
+
+/// Control: with no abort the same late reader gets every group, then the clean end.
+#[tokio::test]
+async fn a_settled_track_delivers_every_group_to_a_late_reader() {
+	assert_whole(round(false, false).await, "arrival order, no abort");
+	assert_whole(round(false, true).await, "sequence order, no abort");
+}