From e2711eb708e65c8ab4ee3de35184c3bcb3fd570e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 03:43:44 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m2/lite-legacy-gap.md Quest-Claim: f7656b52e138c0a4 From 75497ac326e1bd4965e50b67fc4989b7eaf6db38 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 03:54:27 -0700 Subject: [PATCH 2/2] fix(net): lite-05/06 idle copies judge the gap by SUBSCRIBE_START An idle copy on lite-05/06 judged whether its feed skipped past the cache by the first group header. On a descending track that header is the newest group, so readers lost every cached group their max age allowed. Their START names the first group the feed serves, so the copy now goes live once it has both that START and a group header, whichever lands first. No wire or public API change. Co-Authored-By: Claude Opus 5.5 --- quest/m2/README.md | 1 - quest/m2/lite-legacy-gap.md | 24 ------ rs/moq-net/src/lite/subscriber.rs | 109 +++++++++++++++++++++++-- rs/moq-net/tests/integration/rejoin.rs | 68 +++++++++++++++ 4 files changed, 169 insertions(+), 33 deletions(-) delete mode 100644 quest/m2/lite-legacy-gap.md diff --git a/quest/m2/README.md b/quest/m2/README.md index 4f4f2452b0a..f9d90ae7a42 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -93,7 +93,6 @@ lite-07; bitrate enforcement is not part of this release. - [Relay NixOS module](/quest/m2/nix-relay-module.md) - module edits run moq-relay's tests, and `cluster.rootUrl` takes several peers - [Read-only lookup](/quest/m2/read-only-lookup.md) - a live track lookup stops waking the front and every demand watcher - [Ordered owed wait](/quest/m2/ordered-owed-wait.md) - an ordered reader waits within max_delay for a missing group only while the track still owes it -- [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 8dd35ae830f..72aeb18d299 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 { @@ -951,14 +969,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, @@ -2904,6 +2924,8 @@ mod tests { compressed: false, }, tail: Default::default(), + start: None, + first: None, }, ); let mut accept = UniAccept::new(subscriber, early); @@ -2964,6 +2986,8 @@ mod tests { compressed: false, }, tail: Default::default(), + start: None, + first: None, }, ); @@ -3063,6 +3087,8 @@ mod tests { producer: track.clone(), framing, tail: Default::default(), + start: None, + first: None, }, ); @@ -3250,6 +3276,8 @@ mod tests { compressed: false, }, tail: Default::default(), + start: None, + first: None, }, ); @@ -3310,6 +3338,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. @@ -3360,6 +3441,8 @@ mod tests { compressed: false, }, tail: tail.clone(), + start: None, + first: None, }, ); @@ -3838,6 +3921,8 @@ mod tests { compressed: false, }, tail: Default::default(), + start: None, + first: None, }, ); @@ -6142,6 +6227,8 @@ mod tests { compressed: false, }, tail: Default::default(), + start: None, + first: None, }, ); let mut held = Vec::new(); @@ -6845,6 +6932,8 @@ impl TrackServe { producer: producer.clone(), framing: lite::Framing::new(producer.info(), self.subscriber.version), tail: tail.clone(), + start: None, + first: None, }, ); @@ -7945,6 +8034,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 52ad9633df6..27f6f84cd39 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