fix(net): lite-05/06 idle copies judge the gap by SUBSCRIBE_START

quest/m2/README.md

@@ -90,7 +90,6 @@ lite-07; bitrate enforcement is not part of this release.
 - [Shared group serving and ingest in JS](/quest/m2/shared-serve-js.md) - the same split in js/net, following the Rust seam
 - [Paired bench BASE](/quest/m2/bench-base-paired.md) - `just bench <base>` uses the nightly's paired comparison and the old one is deleted
 - [Read-only lookup](/quest/m2/read-only-lookup.md) - a live track lookup stops waking the front and every demand watcher
-- [lite-05/06 gap judgement](/quest/m2/lite-legacy-gap.md) - a resumed copy on lite-05/06 never reads a descending track's newest group as its start
 - [Refill evicted groups](/quest/m2/evicted-refill.md) - a group evicted after a reader judged it but before it took it is refilled, never delivered twice, and guarded against thrash
 - [Receive pool](/quest/m2/receive-pool.md) - one byte budget per worker covers every connection's receive credit and the cache, billing each frame once
 - [Rust-backed @moq/net](/quest/m2/rs2ts/README.md) - measure browser baselines, complete WASM bindings, and adopt Rust behind @moq/net before the later JS-only stage

rs/moq-net/src/lite/subscriber.rs

@@ -100,6 +100,24 @@ struct TrackEntry {
 	framing: lite::Framing,
 	/// The groups received so far, so the subscription's end can wait for the ones owed.
 	tail: kio::Producer<Tail>,
+	/// Lite-05/06: where the feed starts, from SUBSCRIBE_START.
+	start: Option<u64>,
+	/// Lite-05/06: the first group header, which says where the live feed is.
+	first: Option<Position>,
+}
+
+impl TrackEntry {
+	/// Lite-05/06: an idle copy goes live once its answer shows both where the feed starts
+	/// and a group of it. START names the start but not the live edge, and a group can
+	/// overtake it: on a descending track the first to land is the newest, which would read
+	/// as a gap after the cache.
+	fn answered(&mut self, start: Option<u64>, first: Option<Position>) {
+		self.start = self.start.or(start);
+		self.first = self.first.or(first);
+		if let (Some(start), Some(first)) = (self.start, self.first) {
+			self.producer.set_live(Some(first), Some(start));
+		}
+	}
 }
 
 impl Subscriber {
@@ -1033,14 +1051,16 @@ impl GroupRecv {
 			// head gets back. The copy goes live once the cache shows that group.
 			// Lite-07's answer does, and a group can overtake it: on a descending
 			// track the first to land is the newest, not where the feed starts.
-			if !entry.producer.is_live() && !self.subscriber.version.has_largest() {
-				entry.producer.set_live(
-					Some(Position {
-						group: hdr.sequence,
-						frame: hdr.frame_start,
-					}),
-					None,
-				);
+			if !entry.producer.is_live() {
+				let first = Position {
+					group: hdr.sequence,
+					frame: hdr.frame_start,
+				};
+				if !self.subscriber.version.has_track_stream() {
+					entry.producer.set_live(Some(first), None);
+				} else if !self.subscriber.version.has_largest() {
+					entry.answered(None, Some(first));
+				}
 			}
 			let group = match received {
 				Ok(group) => group,
@@ -3002,6 +3022,8 @@ mod tests {
 							compressed: false,
 						},
 						tail: Default::default(),
+						start: None,
+						first: None,
 					},
 				);
 				let mut accept = UniAccept::new(subscriber, early);
@@ -3062,6 +3084,8 @@ mod tests {
 						compressed: false,
 					},
 					tail: Default::default(),
+					start: None,
+					first: None,
 				},
 			);
 
@@ -3161,6 +3185,8 @@ mod tests {
 				producer: track.clone(),
 				framing,
 				tail: Default::default(),
+				start: None,
+				first: None,
 			},
 		);
 
@@ -3348,6 +3374,8 @@ mod tests {
 					compressed: false,
 				},
 				tail: Default::default(),
+				start: None,
+				first: None,
 			},
 		);
 
@@ -3408,6 +3436,59 @@ mod tests {
 		assert!(matches!(res, Err(Error::BoundsExceeded(_))), "{res:?}");
 	}
 
+	/// Lite-05/06: an idle copy judges the gap after its cache by SUBSCRIBE_START, whichever
+	/// lands first. A descending track's newest group overtaking the START waits for it,
+	/// rather than reading as a feed that skipped past the cache.
+	#[test]
+	fn an_idle_copy_judges_the_gap_by_the_start() {
+		for header_first in [true, false] {
+			let mut track = track::Producer::new(
+				std::sync::Arc::new(crate::broadcast::Info::default()),
+				"video",
+				crate::track::Info::timed(),
+			);
+			let mut cached = track.create_group(group::Info { sequence: 0 }).unwrap();
+			cached.write_frame(Timestamp::ZERO, &b"0"[..]).unwrap();
+			cached.finish().unwrap();
+			track.set_idle();
+			let mut entry = TrackEntry {
+				producer: track.clone(),
+				framing: Default::default(),
+				tail: Default::default(),
+				start: None,
+				first: None,
+			};
+
+			// The route resends from the cached head, newest first.
+			if header_first {
+				entry.answered(None, Some(Position::group(3)));
+				assert!(!track.is_live(), "the header waits for the start");
+				entry.answered(Some(0), None);
+			} else {
+				entry.answered(Some(0), None);
+				assert!(!track.is_live(), "the start waits for a group");
+				entry.answered(None, Some(Position::group(3)));
+			}
+			let mut newest = track.create_group(group::Info { sequence: 3 }).unwrap();
+			newest
+				.write_frame(Timestamp::from_millis(300).unwrap(), &b"3"[..])
+				.unwrap();
+			newest.finish().unwrap();
+			assert!(track.is_live(), "header first: {header_first}");
+
+			let subscription = crate::track::Subscription::default()
+				.with_max_delay(std::time::Duration::from_secs(10))
+				.with_range(0..);
+			let mut reader = track.subscribe(subscription);
+			let mut sequences = Vec::new();
+			while let Poll::Ready(Ok(Some(group))) = reader.poll_recv_group(&kio::Waiter::noop()) {
+				sequences.push(group.sequence);
+			}
+			sequences.sort();
+			assert_eq!(sequences, [0, 3], "header first: {header_first}");
+		}
+	}
+
 	/// An idle copy that resubscribes gets its newest cached group again. That stream only
 	/// duplicates the cache, but lite-07's SUBSCRIBE_END still counts it, so the tail must
 	/// too, or the subscription waits out the grace before it ends.
@@ -3458,6 +3539,8 @@ mod tests {
 					compressed: false,
 				},
 				tail: tail.clone(),
+				start: None,
+				first: None,
 			},
 		);
 
@@ -3936,6 +4019,8 @@ mod tests {
 					compressed: false,
 				},
 				tail: Default::default(),
+				start: None,
+				first: None,
 			},
 		);
 
@@ -6287,6 +6372,8 @@ mod tests {
 					compressed: false,
 				},
 				tail: Default::default(),
+				start: None,
+				first: None,
 			},
 		);
 		let mut held = Vec::new();
@@ -7010,6 +7097,8 @@ impl TrackServe {
 				producer: producer.clone(),
 				framing: lite::Framing::new(producer.info(), self.subscriber.version),
 				tail: tail.clone(),
+				start: None,
+				first: None,
 			},
 		);
 
@@ -8118,6 +8207,10 @@ impl ServeLoop {
 											if let Some(largest) = start.largest {
 												self.serving.report_largest(largest);
 											}
+										} else if let Some(entry) =
+											serve.subscriber.subscribes.lock().get_mut(&active.id)
+										{
+											entry.answered(served, None);
 										}
 										// A START describes the demand the SUBSCRIBE carried.
 										// It applies only while the current start still matches

rs/moq-net/tests/integration/rejoin.rs

@@ -181,6 +181,74 @@ async fn rejoin_skips_a_cache_kept_by_another_handle() {
 	}
 }
 
+/// A rejoin keeps the cached group its max age still allows, though the backlog lands
+/// newest first.
+///
+/// A fetch-only reader holds the relay's copy, so it idles with group 0 cached when the
+/// last subscriber leaves. The publisher writes groups 1 to 3 meanwhile, and the rejoin
+/// asks again from the cached head, which a descending track answers newest first. Group
+/// 3 landing first is not a gap after group 0, so a reader asking from group 0 gets it.
+#[moq_clock::test]
+async fn rejoin_keeps_the_cache_when_the_backlog_lands_newest_first() {
+	for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"] {
+		moq_clock::timeout(TEST_TIMEOUT, async {
+			let publisher = produce_origin(1);
+			let relay = produce_origin(2);
+
+			let broadcast = publisher.create_broadcast("bench").unwrap();
+			let track = broadcast.create_track("video", timed()).unwrap();
+			broadcast.announce(Default::default()).unwrap();
+
+			let mut options = MockConnectOptions::new(version.parse::<Version>().unwrap());
+			options.server_publish = Some(publisher.consume());
+			options.client_subscribe = Some(relay.clone());
+			let pair = connect_mock(options).await;
+
+			let consumer = relay.consume();
+			let remote = consumer.follow("bench", None).unwrap().available().await.unwrap();
+			let ts = |ms| Timestamp::from_millis(ms).unwrap();
+			let prefs = || track::Subscription::default().with_max_delay(Duration::from_secs(10));
+
+			// Held for fetches only: it keeps the relay's copy without subscribing.
+			let _held = remote.track("video").unwrap();
+
+			let mut group = track.append_group().unwrap();
+			group.write_frame(ts(0), b"0".as_ref()).unwrap();
+			group.finish().unwrap();
+			let mut sub = remote.track("video").unwrap().subscribe(prefs()).await.unwrap();
+			let mut group = sub.recv_group().await.unwrap().unwrap();
+			assert_eq!(read_all(&mut group).await.unwrap(), [b"0".to_vec()], "{version}");
+			drop((group, sub));
+			crate::support::unwanted(track.demand()).await.unwrap();
+
+			// The publisher moves on while nobody subscribes, all within the max age.
+			for sequence in 1..=3u64 {
+				let mut group = track.append_group().unwrap();
+				group.write_frame(ts(sequence * 100), sequence.to_string()).unwrap();
+				group.finish().unwrap();
+			}
+
+			pair.server_transport.hold_unis();
+			let rejoin = remote.track("video").unwrap().subscribe(prefs().with_range(0..));
+			let release = async {
+				moq_clock::sleep(Duration::from_millis(100)).await;
+				pair.server_transport.release_unis_reversed();
+			};
+			let (sub, ()) = futures::join!(rejoin, release);
+			let mut sub = sub.unwrap();
+			let mut rejoined = Vec::new();
+			// Arrival order, newest first: whatever the reader gets once the backlog settled.
+			while let Ok(group) = moq_clock::timeout(Duration::from_secs(1), sub.recv_group()).await {
+				rejoined.push(group.unwrap().unwrap().sequence);
+			}
+			rejoined.sort();
+			assert_eq!(rejoined, [0, 1, 2, 3], "{version}: the rejoin skipped the cache");
+		})
+		.await
+		.unwrap_or_else(|_| panic!("{version}: timed out"));
+	}
+}
+
 /// A reader rejoining an open group keeps reading it, and leaves the relay holding it whole.
 ///
 /// The client still holds the group's head when it rejoins, so it may ask the relay for