From 36e3e4c0e5c954778248108a3529b808f44a95b4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:07:46 -0700 Subject: [PATCH 1/3] quest: claim follow-gap From de1bf476dab34d7a3bc048e308199ec3db8f4054 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:21:17 -0700 Subject: [PATCH 2/3] fix(moq-net): a followed path reports a gap onto a same-epoch prefix The announce cursor delivers by prefix, so a covering prefix's start can arrive ahead of the serving path's end even when it was announced after it. The follower folded the pair into an Update, hiding the gap that ended any request on the old route. A covering route that started since the cursor last ran dry now takes over as an End then a Start, matching JS. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 7 ++++ quest/m0/broadcast-epoch/README.md | 1 - quest/m0/broadcast-epoch/follow-gap.md | 34 ---------------- rs/moq-net/src/model/follow.rs | 55 ++++++++++++++++++++++---- 4 files changed, 55 insertions(+), 42 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/follow-gap.md diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index f790460e9d..f7d5df1887 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -445,6 +445,13 @@ test("follow keeps a gap between routes of one epoch", async () => { expect(await follow.next()).toMatchObject({ kind: "end", prefix: Path.from("pool/job") }); expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool") }); + // A prefix standing while the path's route goes takes over in place. + const again = origin.createBroadcast(Path.from("pool/job")); + again.announce({ epoch }); + expect(await follow.next()).toMatchObject({ kind: "update", prefix: Path.from("pool/job") }); + again.close(); + expect(await follow.next()).toMatchObject({ kind: "update", prefix: Path.from("pool") }); + follow.close(); pool.close(); origin.close(); diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 8652959cd7..4af74ffe27 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -101,5 +101,4 @@ maintainer's pre-release merge of `release` into `main` keeps `main`'s - [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one - [moqsrc](/quest/m0/broadcast-epoch/moqsrc.md) - moqsrc switches to the new broadcast on a `Restart`, keeping its pads by rendition name -- [Follow gap](/quest/m0/broadcast-epoch/follow-gap.md) - a followed path reports a gap that ended its request, even onto a same-epoch covering prefix - [Stats totals and prefix tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and on-demand prefix tracks (decided 2026-10-05) diff --git a/quest/m0/broadcast-epoch/follow-gap.md b/quest/m0/broadcast-epoch/follow-gap.md deleted file mode 100644 index 638676535c..0000000000 --- a/quest/m0/broadcast-epoch/follow-gap.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] A followed path reports a gap that ended its request - -## Goal - -`origin::Consumer::follow` reports a gap that ended the request (an `End` -then `Start`, or a `Restart`) even when the next serving route is a covering -prefix with the same epoch, so `moq play` never misses a handoff while it is -still draining the old run. - -## Plan - -Found by the OpenAI review of #5154 (2026-10-10), which added `follow` in -moq-net and `@moq/net`. A same-epoch handoff from the exact route to a -covering prefix, after a gap that ended the old request, reaches the Rust -follower as an `Update`, because the announce cursor delivers by prefix and -the follower never sees the gap. If that `Update` arrives while `moq play` is -draining the old run, playback never restarts. JS sees the gap and is -correct. - -Decided 2026-10-10: merge #5154 and fix the cause in moq-net here, in m0 -because the epoch line gates the release. Rejected: re-requesting in -`moq play` whenever a run fails while the path is still served, which loops -tightly on a broadcast that fails every time. - -Fix it in the announce cursor or the follower so the gap surfaces, matching -JS. Regression with simulated time: an exact route ends, its request ends, -then a same-epoch prefix route takes over while a reader is still draining; -the follower must report the handoff, and `moq play` must restart. Leave the -follower unpolled until the replacement is announced, since awaiting its -`End` first avoids the bug and passes on the broken code, as -`follow_keeps_a_gap_between_routes_of_one_epoch` does today. - -Public API: no new items, but `follow` reports a gap where it reported an -`Update`, a behavior change to call out in the PR. Wire: none. diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index f8dd57ddc4..4204f0c59f 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -17,6 +17,11 @@ pub struct Follow { path: PathOwned, /// Every route standing over the path. covering: Vec, + /// Covering prefixes that started since the cursor last ran dry. The cursor delivers by + /// prefix, not in order, so one of these may have arrived after the serving route ended. + fresh: Vec, + /// Nothing has been reported serving the path since the last end. + idle: bool, } impl Follow { @@ -25,6 +30,8 @@ impl Follow { announced, path, covering: Vec::new(), + fresh: Vec::new(), + idle: true, } } @@ -40,11 +47,12 @@ impl Follow { /// beneath it) starts on the route serving the path rather than starting and restarting. /// Once something serves it, each change comes through on its own: an end followed by a /// start is a gap that ended any request on the old route, even when both routes carry one - /// epoch. + /// epoch. A route that started alongside the serving route's end may have arrived after it, + /// so taking over from it is reported as that gap. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - if self.serving().is_none() { + if self.idle { let ended = loop { - match self.announced.poll_next(waiter) { + match self.poll_announced(waiter) { Poll::Ready(Some(event)) => { self.fold(event); } @@ -53,22 +61,36 @@ impl Follow { } }; return match (self.serving().cloned(), ended) { - (Some(serving), _) => Poll::Ready(Some(AnnounceEvent::Start(serving))), + (Some(serving), _) => { + self.idle = false; + Poll::Ready(Some(AnnounceEvent::Start(serving))) + } (None, true) => Poll::Ready(None), (None, false) => Poll::Pending, }; } loop { - let Some(event) = ready!(self.announced.poll_next(waiter)) else { + let Some(event) = ready!(self.poll_announced(waiter)) else { return Poll::Ready(None); }; if let Some(event) = self.fold(event) { + self.idle = matches!(event, AnnounceEvent::End(_)); return Poll::Ready(Some(event)); } } } + /// The cursor's next event. Running dry means every covering route started before any + /// end still to come, so none is fresh any more. + fn poll_announced(&mut self, waiter: &kio::Waiter) -> Poll> { + let poll = self.announced.poll_next(waiter); + if poll.is_pending() { + self.fresh.clear(); + } + poll + } + /// The most specific route covering the path, which is the one a request resolves. fn serving(&self) -> Option<&Announce> { self.covering @@ -91,7 +113,12 @@ impl Follow { self.covering.retain(|standing| standing.prefix != prefix); let restart = match event { AnnounceEvent::End(_) => false, - AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) => { + AnnounceEvent::Start(announce) => { + self.fresh.push(prefix.clone()); + self.covering.push(announce); + false + } + AnnounceEvent::Update(announce) => { self.covering.push(announce); false } @@ -106,6 +133,12 @@ impl Follow { (None, None) => None, (None, Some(after)) => Some(AnnounceEvent::Start(after)), (Some(before), None) => Some(AnnounceEvent::End(before)), + // The serving route ended, and the one left may have only arrived after it. + (Some(before), Some(after)) + if before.prefix != after.prefix && before.prefix == prefix && self.fresh.contains(&after.prefix) => + { + Some(AnnounceEvent::End(before)) + } (Some(before), Some(after)) if before.prefix != after.prefix => { let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; Some(match same { @@ -217,10 +250,18 @@ mod tests { let mut follow = origin.consume().follow("pool/job").unwrap(); assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + // Both land before the follower reads again, and the cursor delivers the prefix's start + // ahead of the path's end, but the path went unserved in between. drop(exact); - assert_eq!(followed(&mut follow).await, ("end", "pool/job".into())); let _pool = origin.dynamic("pool", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("end", "pool/job".into())); assert_eq!(followed(&mut follow).await, ("start", "pool".into())); + + // A prefix standing while the path's route goes takes over in place. + let exact = origin.publish("pool/job", route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("update", "pool/job".into())); + drop(exact); + assert_eq!(followed(&mut follow).await, ("update", "pool".into())); } /// Without an epoch nothing says two routes serve the same bytes. From b3948dbb86841c1d369932b09017fb0cec6d7d0f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:37:35 -0700 Subject: [PATCH 3/3] fix(moq-net): a prefix seen before the serving route's last event hands over in place Taking any event for a prefix proves no end was queued there yet, so a covering route whose start was taken before the serving route's latest event stood before that route's end. Track take sequence numbers instead of a fresh set that only cleared when the cursor ran dry, which reported a spurious gap after a returned Restart or Update. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 20 ++++++ rs/moq-net/src/model/follow.rs | 124 ++++++++++++++++++++++----------- 2 files changed, 102 insertions(+), 42 deletions(-) diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index f7d5df1887..05313d4367 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -457,6 +457,26 @@ test("follow keeps a gap between routes of one epoch", async () => { origin.close(); }); +test("follow hands over to a prefix that started before the last change", async () => { + const origin = new Producer(); + const exact = origin.createBroadcast(Path.from("pool/job")); + exact.announce({ epoch: Epoch.mint() }); + const follow = origin.consume().follow(Path.from("pool/job")); + expect(await follow.next()).toMatchObject({ kind: "start", prefix: Path.from("pool/job") }); + + const epoch = Epoch.mint(); + const pool = origin.dynamic(Path.from("pool"), { epoch }); + exact.announce({ epoch }); + expect(await follow.next()).toMatchObject({ kind: "restart", prefix: Path.from("pool/job") }); + + exact.close(); + expect(await follow.next()).toMatchObject({ kind: "update", prefix: Path.from("pool") }); + + follow.close(); + pool.close(); + origin.close(); +}); + test("follow refuses a path no pattern can spell", () => { const origin = new Producer(); expect(() => origin.consume().follow(Path.from("camera*main"))).toThrow(); diff --git a/rs/moq-net/src/model/follow.rs b/rs/moq-net/src/model/follow.rs index 4204f0c59f..879210e16b 100644 --- a/rs/moq-net/src/model/follow.rs +++ b/rs/moq-net/src/model/follow.rs @@ -16,21 +16,36 @@ pub struct Follow { announced: AnnounceConsumer, path: PathOwned, /// Every route standing over the path. - covering: Vec, - /// Covering prefixes that started since the cursor last ran dry. The cursor delivers by - /// prefix, not in order, so one of these may have arrived after the serving route ended. - fresh: Vec, + covering: Vec, + /// How many events the cursor has delivered. + taken: u64, + /// What `taken` was when the cursor last ran dry, so held no end. + dry: u64, /// Nothing has been reported serving the path since the last end. idle: bool, } +/// A route standing over the followed path. +/// +/// The cursor delivers by prefix, not in order, so a route's end may have happened before +/// another prefix's start that it delivers first. Taking an event for a prefix proves no end +/// was queued there yet, so any route that started before that take stood before its end. +struct Standing { + announce: Announce, + /// When its start was taken. + since: u64, + /// When its latest event was taken. + seen: u64, +} + impl Follow { pub(super) fn new(announced: AnnounceConsumer, path: PathOwned) -> Self { Self { announced, path, covering: Vec::new(), - fresh: Vec::new(), + taken: 0, + dry: 0, idle: true, } } @@ -47,8 +62,9 @@ impl Follow { /// beneath it) starts on the route serving the path rather than starting and restarting. /// Once something serves it, each change comes through on its own: an end followed by a /// start is a gap that ended any request on the old route, even when both routes carry one - /// epoch. A route that started alongside the serving route's end may have arrived after it, - /// so taking over from it is reported as that gap. + /// epoch. A route whose start came after the serving route's last event, with the cursor + /// never running dry in between, may have arrived after its end, so taking over from it is + /// reported as that gap. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { if self.idle { let ended = loop { @@ -60,7 +76,7 @@ impl Follow { Poll::Pending => break false, } }; - return match (self.serving().cloned(), ended) { + return match (self.serving().map(|standing| standing.announce.clone()), ended) { (Some(serving), _) => { self.idle = false; Poll::Ready(Some(AnnounceEvent::Start(serving))) @@ -81,21 +97,22 @@ impl Follow { } } - /// The cursor's next event. Running dry means every covering route started before any - /// end still to come, so none is fresh any more. + /// The cursor's next event, counting what it delivers and when it runs dry. fn poll_announced(&mut self, waiter: &kio::Waiter) -> Poll> { let poll = self.announced.poll_next(waiter); - if poll.is_pending() { - self.fresh.clear(); + match poll { + Poll::Ready(Some(_)) => self.taken += 1, + Poll::Pending => self.dry = self.taken, + Poll::Ready(None) => {} } poll } /// The most specific route covering the path, which is the one a request resolves. - fn serving(&self) -> Option<&Announce> { + fn serving(&self) -> Option<&Standing> { self.covering .iter() - .max_by_key(|announce| announce.prefix.as_str().len()) + .max_by_key(|standing| standing.announce.prefix.as_str().len()) } /// Apply one route's event, returning what it means for the path, if anything. @@ -108,38 +125,43 @@ impl Follow { return None; } let prefix = announce.prefix.clone(); - let before = self.serving().cloned(); - - self.covering.retain(|standing| standing.prefix != prefix); - let restart = match event { - AnnounceEvent::End(_) => false, - AnnounceEvent::Start(announce) => { - self.fresh.push(prefix.clone()); - self.covering.push(announce); - false - } - AnnounceEvent::Update(announce) => { - self.covering.push(announce); - false - } - AnnounceEvent::Restart(announce) => { - self.covering.push(announce); - true - } + let before = self + .serving() + .map(|standing| (standing.announce.clone(), standing.seen)); + + let standing = self + .covering + .iter() + .position(|standing| standing.announce.prefix == prefix) + .map(|index| self.covering.swap_remove(index).since); + let (announce, since, restart) = match event { + AnnounceEvent::End(_) => (None, None, false), + AnnounceEvent::Start(announce) => (Some(announce), None, false), + AnnounceEvent::Update(announce) => (Some(announce), standing, false), + AnnounceEvent::Restart(announce) => (Some(announce), standing, true), }; + if let Some(announce) = announce { + self.covering.push(Standing { + announce, + since: since.unwrap_or(self.taken), + seen: self.taken, + }); + } - let after = self.serving().cloned(); + let after = self + .serving() + .map(|standing| (standing.announce.clone(), standing.since)); match (before, after) { (None, None) => None, - (None, Some(after)) => Some(AnnounceEvent::Start(after)), - (Some(before), None) => Some(AnnounceEvent::End(before)), - // The serving route ended, and the one left may have only arrived after it. - (Some(before), Some(after)) - if before.prefix != after.prefix && before.prefix == prefix && self.fresh.contains(&after.prefix) => + (None, Some((after, _))) => Some(AnnounceEvent::Start(after)), + (Some((before, _)), None) => Some(AnnounceEvent::End(before)), + // The serving route ended, and the one left may have started after it did. + (Some((before, seen)), Some((after, since))) + if before.prefix == prefix && before.prefix != after.prefix && since > self.dry.max(seen) => { Some(AnnounceEvent::End(before)) } - (Some(before), Some(after)) if before.prefix != after.prefix => { + (Some((before, _)), Some((after, _))) if before.prefix != after.prefix => { let same = before.route.epoch.is_some() && before.route.epoch == after.route.epoch; Some(match same { true => AnnounceEvent::Update(after), @@ -147,9 +169,9 @@ impl Follow { }) } // A route less specific than the serving one changed, which no request sees. - (Some(_), Some(after)) if after.prefix != prefix => None, - (Some(_), Some(after)) if restart => Some(AnnounceEvent::Restart(after)), - (Some(_), Some(after)) => Some(AnnounceEvent::Update(after)), + (Some(_), Some((after, _))) if after.prefix != prefix => None, + (Some(_), Some((after, _))) if restart => Some(AnnounceEvent::Restart(after)), + (Some(_), Some((after, _))) => Some(AnnounceEvent::Update(after)), } } } @@ -264,6 +286,24 @@ mod tests { assert_eq!(followed(&mut follow).await, ("update", "pool".into())); } + /// A prefix that started before the serving route's last change stood before the route's + /// end, so it takes over in place even when the cursor never ran dry in between. + #[moq_net_sim::test] + async fn follow_hands_over_to_a_prefix_that_started_before_the_last_change() { + let origin = Hop::new(1).unwrap().produce(); + let exact = origin.publish("pool/job", route(&Epoch::mint())).unwrap(); + let mut follow = origin.consume().follow("pool/job").unwrap(); + assert_eq!(followed(&mut follow).await, ("start", "pool/job".into())); + + let epoch = Epoch::mint(); + let _pool = origin.dynamic("pool", route(&epoch)).unwrap(); + exact.announce(route(&epoch)).unwrap(); + assert_eq!(followed(&mut follow).await, ("restart", "pool/job".into())); + + drop(exact); + assert_eq!(followed(&mut follow).await, ("update", "pool".into())); + } + /// Without an epoch nothing says two routes serve the same bytes. #[moq_net_sim::test] async fn follow_restarts_onto_a_more_specific_route_without_an_epoch() {