fix(net): a group withheld for its first frame holds the end

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

@@ -1348,12 +1348,19 @@ impl TrackState {
 		// `sealed` also ends a locally closed receive track without a declared end.
 		// An abort still wins unless that end had already settled: a group below it was still open.
 		let reached = self.sealed
-			|| self
-				.final_sequence
-				.is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin);
+			|| self.final_sequence.is_some_and(|fin| {
+				self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin && !self.withholds(fin)
+			});
 		reached && (self.abort.is_none() || self.settled)
 	}
 
+	/// Whether a group below `fin` is still withheld from readers until its first frame
+	/// lands (see [`Producer::receive_group`]): the end is not reached before it shows, or
+	/// readers would end without it.
+	fn withholds(&self, fin: u64) -> bool {
+		self.lookup.range(..fin).any(|(_, slot)| slot.pending)
+	}
+
 	/// Whether the declared end is reached and every cached group below it finished,
 	/// so nothing the end promised is still in flight.
 	fn is_settled(&self) -> bool {
@@ -6448,6 +6455,72 @@ mod test {
 		assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
 	}
 
+	/// A route can declare the end, then land every group's header before its first frame.
+	/// Those groups are withheld until their frames land, so the end is not reached before
+	/// they show: a reader ending at the boundary would never see them.
+	#[test]
+	fn a_withheld_group_holds_the_end() {
+		let mut producer = track_producer("test", None);
+		let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(60)));
+		producer.finish_at(2).unwrap();
+		let mut received: Vec<_> = (0..2)
+			.map(|sequence| producer.receive_group(group::Info { sequence }).unwrap())
+			.collect();
+		assert!(
+			subscriber.recv_group().now_or_never().is_none(),
+			"the end waits for the withheld groups"
+		);
+
+		for group in &mut received {
+			group
+				.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
+				.unwrap();
+			producer.reveal_group(group);
+		}
+		assert_eq!(subscriber.assert_group().sequence, 0);
+		assert_eq!(subscriber.assert_group().sequence, 1);
+		assert!(subscriber.recv_group().now_or_never().unwrap().unwrap().is_none());
+	}
+
+	/// A withheld group whose stream ends with no frame, cleanly or reset, shows when it
+	/// ends, which wakes a reader parked on the end.
+	#[moq_net_sim::test]
+	async fn a_withheld_group_that_ends_empty_releases_the_end() {
+		for abort in [false, true] {
+			let mut producer = track_producer("test", None);
+			let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(60)));
+			producer.finish_at(1).unwrap();
+			let group = producer.receive_group(group::Info { sequence: 0 }).unwrap();
+			let handle = group.clone();
+
+			let first = {
+				let mut next = std::pin::pin!(subscriber.recv_group());
+				assert!(
+					futures::poll!(next.as_mut()).is_pending(),
+					"the end waits for the group"
+				);
+				match abort {
+					true => group.abort(Error::Cancel).unwrap(),
+					false => group.finish().unwrap(),
+				}
+				producer.reveal_group(&handle);
+				moq_net_sim::timeout(Duration::from_secs(1), next)
+					.await
+					.expect("the reveal wakes the reader")
+					.unwrap()
+			};
+
+			match abort {
+				// A reset group shows nothing, so the reader ends.
+				true => assert!(first.is_none()),
+				false => {
+					assert_eq!(first.expect("the empty group").sequence, 0);
+					assert!(subscriber.recv_group().now_or_never().unwrap().unwrap().is_none());
+				}
+			}
+		}
+	}
+
 	/// The ordinary live case, at the default real-time budget: 2s GOPs produced one at
 	/// a time and read as they arrive. The budget must take the live edge without
 	/// shortening the group the reader is already on, so every frame of every group is

rs/moq-net/tests/track_tail.rs

@@ -358,3 +358,63 @@ async fn a_lost_datagram_never_delays_the_end() {
 		drop((pair, _keep));
 	}
 }
+
+/// A publisher can declare the end, then open each group's stream before writing its first
+/// frame. The subscriber withholds a group until that frame lands, and the end waits for
+/// it: a reader that ended at the boundary would miss every group.
+#[moq_net_sim::test]
+async fn a_group_whose_first_frame_trails_its_header_is_delivered() {
+	for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"] {
+		let Pair {
+			pair,
+			mut track,
+			remote,
+			_keep,
+		} = connect(version).await;
+
+		let reader = moq_net_sim::spawn(async move {
+			let subscription = moq_net::track::Subscription::default()
+				.with_start(moq_net::track::Position::group(0))
+				.with_max_delay(TIMEOUT);
+			let mut sub = remote
+				.track("video")
+				.unwrap()
+				.subscribe(subscription)
+				.await
+				.expect("subscribe");
+			let mut groups = Vec::new();
+			while let Some(mut group) = sub.recv_group().await.expect("track aborted") {
+				let frame = group.read_frame().await.expect("group aborted").expect("a frame");
+				assert_eq!(&frame.payload[..], PAYLOAD);
+				groups.push(group.sequence);
+			}
+			groups
+		});
+		moq_net_sim::timeout(TIMEOUT, support::harness::subscribed(&track))
+			.await
+			.expect("no subscriber appeared");
+
+		track.finish_at(2).unwrap();
+		let mut groups = [track.append_group().unwrap(), track.append_group().unwrap()];
+		// Paused time advances only once every task is idle: both headers and the end
+		// have reached the subscriber, ahead of any frame.
+		moq_net_sim::sleep(GRACE / 10).await;
+		assert!(
+			!reader.is_finished(),
+			"{version}: the track ended before its groups showed"
+		);
+
+		for group in &mut groups {
+			group.write_frame(Timestamp::ZERO, PAYLOAD).unwrap();
+			group.finish().unwrap();
+		}
+		let mut groups = moq_net_sim::timeout(TIMEOUT, reader)
+			.await
+			.expect("the subscription never ended")
+			.expect("reader panicked");
+		// Arrival order: either group's frame may land first.
+		groups.sort();
+		assert_eq!(groups, [0, 1], "{version}");
+		drop((track, pair, _keep));
+	}
+}