diff --git a/quest/m2/README.md b/quest/m2/README.md
index 77c2b2ecdc6..a83175ce1e7 100644
--- a/quest/m2/README.md
+++ b/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 ` 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
diff --git a/quest/m2/lite-legacy-gap.md b/quest/m2/lite-legacy-gap.md
deleted file mode 100644
index c52f5485ff0..00000000000
--- a/quest/m2/lite-legacy-gap.md
+++ /dev/null
@@ -1,24 +0,0 @@
-# [S] Resumed copies judge the gap on lite-05/06
-
-## Goal
-
-On moq-lite-05 and 06, an idle resumed copy never misreads a descending
-track's newest group as where its feed starts, so readers keep the cached
-groups their max age still allows, as they do on lite-07.
-
-## Plan
-
-From [moq-dev/moq#6183](https://github.com/moq-dev/moq/pull/6183), which
-fixed lite-07 by judging the gap from SUBSCRIBE_START's first group. On
-05/06 the copy still judges it by the first group header, and on a
-descending track the first header to land can be the newest group. Their
-START has no `largest`, so this needs its own small design within those
-versions' wire, which is published and must not change. Propose it in the
-PR with options and a recommendation; leaving 05/06 as is, documented, is
-one option.
-
-Public API: none. Wire: none.
-
-## Related
-
-- [Resume through an outage](https://github.com/moq-dev/moq/pull/6183) - the lite-07 fix
diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs
index b905152d32e..df50ec77974 100644
--- a/rs/moq-net/src/lite/subscriber.rs
+++ b/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,
+ /// Lite-05/06: where the feed starts, from SUBSCRIBE_START.
+ start: Option,
+ /// Lite-05/06: the first group header, which says where the live feed is.
+ first: Option,
+}
+
+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, first: Option) {
+ 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
diff --git a/rs/moq-net/tests/integration/rejoin.rs b/rs/moq-net/tests/integration/rejoin.rs
index 0f124a2f996..6e10b19851f 100644
--- a/rs/moq-net/tests/integration/rejoin.rs
+++ b/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::().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