From 4a44882f526169544d6bd827ad35a7f9fb87c7b4 Mon Sep 17 00:00:00 2001 From: Thomas Drapier Date: Sat, 10 Oct 2026 11:48:05 +0100 Subject: [PATCH 1/4] fix(net): don't renew a prefix whose winner already restarted Co-Authored-By: Claude Opus 5.5 Co-authored-by: Cursor --- rs/moq-net/src/model/origin.rs | 28 ++++++++++++++++++++++++---- rs/moq-net/tests/restart.rs | 11 +++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 659e73d57d..5300d76c15 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2658,18 +2658,21 @@ async fn serve_front(task: FrontTask) -> Vec { let Some(instance) = instances.get(&source).cloned() else { continue; }; - let prefix = { + let (prefix, winner) = { let table = shared.read(); - table + let prefix = table .routes .covering(&path.as_path()) .find(|entry| entry.id == route) - .map(|entry| entry.prefix.clone()) + .map(|entry| entry.prefix.clone()); + let winner = prefix.as_ref().and_then(|prefix| table.presented(prefix, horizon)); + (prefix, winner) }; resolved = Some(Resolution { instance: instance.clone(), route, prefix, + winner, renewed: false, }); if let Ok(mut pending) = request.write() @@ -3678,6 +3681,16 @@ impl OriginState { } best } + + /// The instance announce cursors behind `horizon` present at `prefix`: its best + /// route there, ranked as [`Self::sync_route`] ranks them. + fn presented(&self, prefix: &Path, horizon: Horizon) -> Option { + self.routes + .at(prefix) + .filter(|entry| entry.live() && horizon.admits(entry)) + .min_by_key(|entry| route_order(prefix, entry)) + .map(RouteEntry::instance) + } } /// One-shot result of a dynamic broadcast request. @@ -3707,6 +3720,8 @@ struct Resolution { route: u64, /// That route's prefix, when it was still in the table as the front resolved. prefix: Option, + /// The instance the prefix's announcements presented as the front resolved. + winner: Option, /// Whether the front already renewed the prefix; see [`Self::moved`]. renewed: bool, } @@ -3715,7 +3730,9 @@ impl Resolution { /// The prefix to [renew](OriginState::renew), once per front: on a route without /// an epoch, a request for the path, beneath the prefix, now resolves through /// another route there. The prefix's own winner may be unchanged, so nothing else - /// tells its announce cursors that what they resolved under it moved. + /// tells its announce cursors that what they resolved under it moved. Once another + /// instance wins the prefix, they restarted it already, and a renewal would restart + /// it again. fn moved(&mut self, table: &OriginState, path: &Path, horizon: Horizon) -> Option { if self.renewed || !matches!(self.instance, Instance::Route(_)) { return None; @@ -3729,6 +3746,9 @@ impl Resolution { if best.id == self.route || best.prefix != *prefix || best.epoch.is_some() { return None; } + if table.presented(prefix, horizon) != self.winner { + return None; + } self.renewed = true; Some(prefix.clone()) } diff --git a/rs/moq-net/tests/restart.rs b/rs/moq-net/tests/restart.rs index ad9d8085a8..f5563210d3 100644 --- a/rs/moq-net/tests/restart.rs +++ b/rs/moq-net/tests/restart.rs @@ -60,6 +60,13 @@ async fn expect_restart(version: Version, announced: &mut announce::Consumer, pr } } +/// Nothing more arrives, past the update hold. +async fn quiet(version: Version, announced: &mut announce::Consumer) { + if let Ok(event) = moq_net_sim::timeout(Duration::from_secs(1), announced.next()).await { + panic!("{version}: expected nothing more, got {event:?}"); + } +} + /// Write one group of one frame tagged `tag`. fn write(track: &track::Producer, tag: &str) { let mut group = track.append_group().unwrap(); @@ -324,6 +331,10 @@ async fn dynamic_epoch(version: &str, from: Option, to: Option) { } claim.write(1, "new:0"); assert_eq!(read(&mut subscription).await, "new:0", "{version}"); + + // The sticky subscription's path resolving through the new route is the restart + // already delivered, not another. + quiet(version, &mut announced).await; } /// The epoch the claim starts at, and a newer one it moves to. From c60a3e471ead61f72f01efa87aa3ef69178635e1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:29:35 -0700 Subject: [PATCH 2/4] fix(net): suppress a renewal only when every cursor restarted since the last join The prefix winner a front snapshotted ignored cursor scopes, so a route a scoped cursor cannot see changing the overall winner suppressed a renewal that cursor needed. Ask the cursors instead: renew unless every cursor presenting the prefix presents an instance minted after a requester last joined the front. Counting from the last join, not the resolution, covers a requester that re-requested after a restart and joined the same front. Once settled for a join, the move stays settled, so a winner moving back to an older route adds no restart. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 210 +++++++++++++++++++++++++++------ 1 file changed, 174 insertions(+), 36 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 5300d76c15..c4a42df128 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -989,6 +989,8 @@ struct RemoteFront { /// The front's broadcast, weak: dead once the front ends, so a /// later request re-creates the front instead of joining a corpse. broadcast: broadcast::WeakConsumer, + /// [`OriginState::next_route`] as a requester last joined; see [`Resolution::moved`]. + joined: Arc, } /// The last route a cursor observed: the instance it serves, its metadata, @@ -2264,6 +2266,8 @@ struct FrontTask { watch: Watch, /// Resolves the requesters parked on the front's channel. request: kio::Producer, + /// See [`RemoteFront::joined`]. + joined: Arc, timers: Clock, } @@ -2409,6 +2413,7 @@ async fn serve_front(task: FrontTask) -> Vec { horizon, watch, request, + joined, timers, } = task; @@ -2482,9 +2487,10 @@ async fn serve_front(task: FrontTask) -> Vec { .serving() .and_then(|id| sources.get(&id)) .is_some_and(|source| source.is_closing()); + let joined = joined.load(Ordering::Acquire); let renew = resolved .as_mut() - .and_then(|resolved| resolved.moved(&table, &path.as_path(), horizon)); + .and_then(|resolved| resolved.moved(&table, &path.as_path(), horizon, joined)); drop(table); if let Some(prefix) = renew { shared.lock().renew(&prefix); @@ -2658,22 +2664,17 @@ async fn serve_front(task: FrontTask) -> Vec { let Some(instance) = instances.get(&source).cloned() else { continue; }; - let (prefix, winner) = { - let table = shared.read(); - let prefix = table - .routes - .covering(&path.as_path()) - .find(|entry| entry.id == route) - .map(|entry| entry.prefix.clone()); - let winner = prefix.as_ref().and_then(|prefix| table.presented(prefix, horizon)); - (prefix, winner) - }; + let prefix = shared + .read() + .routes + .covering(&path.as_path()) + .find(|entry| entry.id == route) + .map(|entry| entry.prefix.clone()); resolved = Some(Resolution { instance: instance.clone(), route, prefix, - winner, - renewed: false, + settled: None, }); if let Ok(mut pending) = request.write() && pending.resolved.is_none() @@ -3682,14 +3683,30 @@ impl OriginState { best } - /// The instance announce cursors behind `horizon` present at `prefix`: its best - /// route there, ranked as [`Self::sync_route`] ranks them. - fn presented(&self, prefix: &Path, horizon: Horizon) -> Option { - self.routes + /// Whether every cursor presenting `prefix` restarted it since route generation + /// `since`: none still presents an instance without an epoch from before, which + /// is what a [renewal](Self::renew) would restart. + fn restarted_since(&self, prefix: &Path, since: u64) -> bool { + let old = |instance: &Instance| matches!(instance, Instance::Route(generation) if *generation < since); + // Without a route there from before, no cursor can present one: the common + // case once a renewal moved every generation, and cheaper than the cursors. + let Some(claim) = self + .routes .at(prefix) - .filter(|entry| entry.live() && horizon.admits(entry)) - .min_by_key(|entry| route_order(prefix, entry)) - .map(RouteEntry::instance) + .find(|entry| old(&entry.instance())) + .map(|entry| &entry.claim) + else { + return true; + }; + self.routes.cursors_touching(prefix).into_iter().all(|id| { + let Some(cursor) = self.cursors.get(&id) else { + return true; + }; + let current = cursor + .presented(prefix, claim) + .and_then(|presented| cursor.current.get(&presented)); + !current.is_some_and(|(instance, ..)| old(instance)) + }) } } @@ -3720,21 +3737,22 @@ struct Resolution { route: u64, /// That route's prefix, when it was still in the table as the front resolved. prefix: Option, - /// The instance the prefix's announcements presented as the front resolved. - winner: Option, - /// Whether the front already renewed the prefix; see [`Self::moved`]. - renewed: bool, + /// The [`RemoteFront::joined`] the prefix last restarted for the path's move, by + /// a renewal or on its own; see [`Self::moved`]. + settled: Option, } impl Resolution { - /// The prefix to [renew](OriginState::renew), once per front: on a route without - /// an epoch, a request for the path, beneath the prefix, now resolves through - /// another route there. The prefix's own winner may be unchanged, so nothing else - /// tells its announce cursors that what they resolved under it moved. Once another - /// instance wins the prefix, they restarted it already, and a renewal would restart - /// it again. - fn moved(&mut self, table: &OriginState, path: &Path, horizon: Horizon) -> Option { - if self.renewed || !matches!(self.instance, Instance::Route(_)) { + /// The prefix to [renew](OriginState::renew): on a route without an epoch, a + /// request for the path, beneath the prefix, now resolves through another route + /// there. The prefix's own winner may be unchanged, so nothing else tells its + /// announce cursors that what they resolved under it moved. + /// + /// A cursor that restarted since the last requester `joined` the front already + /// told everyone on it, and a renewal would restart it again. Settled once per + /// join: a renewal leaves the front's instance behind, so nobody joins it after. + fn moved(&mut self, table: &OriginState, path: &Path, horizon: Horizon, joined: u64) -> Option { + if self.settled == Some(joined) || !matches!(self.instance, Instance::Route(_)) { return None; } let prefix = self.prefix.as_ref()?; @@ -3746,11 +3764,11 @@ impl Resolution { if best.id == self.route || best.prefix != *prefix || best.epoch.is_some() { return None; } - if table.presented(prefix, horizon) != self.winner { - return None; + self.settled = Some(joined); + match table.restarted_since(prefix, joined) { + true => None, + false => Some(prefix.clone()), } - self.renewed = true; - Some(prefix.clone()) } } @@ -4682,6 +4700,7 @@ impl Consumer { .then(|| front.broadcast.consume()) .filter(|held| !held.is_closed()); if let Some(held) = held { + front.joined.store(state.next_route, Ordering::Release); let pending = Requesting::queued(front.request.consume(), held) .with_path(requested) .with_stats(scope) @@ -4706,11 +4725,13 @@ impl Consumer { // The front starts held by this request; see [`Front::new`]. let held = broadcast.consume(); let watch = state.watch(&self.shared, &absolute); + let joined = Arc::new(AtomicU64::new(state.next_route)); state.fronts.insert( key, RemoteFront { request: request.clone(), broadcast: held.weak(), + joined: joined.clone(), }, ); // Released before the push: a set whose handles are gone drops the task, @@ -4724,6 +4745,7 @@ impl Consumer { horizon, watch, request, + joined, timers: self.timers.clone(), }, self.tasks.clone(), @@ -5830,6 +5852,121 @@ mod tests { assert!(!replaced.is_clone(&resolved)); } + /// An origin delivering restarts at once, and a way to claim `prefix` with a + /// producer scoped to `scope`, from `hop` at `cost`. + fn scoped_pool() -> (Producer, impl Fn(&str, &str, u64, u64) -> Dynamic) { + let producer = Config { + update_hold: Duration::ZERO, + ..Config::new(origin(1)) + } + .produce(); + let claims = producer.clone(); + let claim = move |prefix: &str, scope: &str, hop: u64, cost: u64| { + claims + .scope("", &scopes(&[scope])) + .unwrap() + .dynamic(prefix, Route::default().with_hops(hops(&[hop])).with_cost(cost)) + .unwrap() + }; + (producer, claim) + } + + /// Request `path` from `consumer` and have `server` answer it: the answer, and + /// the requester's broadcast. + async fn hold(consumer: &Consumer, path: &str, server: &Dynamic) -> (broadcast::Producer, broadcast::Consumer) { + let pending = consumer.request_broadcast(path, None); + let answer = broadcast::Info::new().produce(); + queued(server).await.accept(&answer); + (answer, pending.await.expect("the request resolves")) + } + + /// Let the fronts react, then expect nothing more on `announced`. + async fn quiet(announced: &mut AnnounceConsumer) { + for _ in 0..20 { + moq_net_sim::yield_now().await; + } + announced.assert_next_wait(); + } + + /// Let the fronts react, then expect a restart of `prefix` on `announced`. + async fn restarted(announced: &mut AnnounceConsumer, prefix: &str) -> Route { + for _ in 0..20 { + moq_net_sim::yield_now().await; + } + announced.assert_next_restarted(prefix) + } + + /// A held path moving restarts a scoped cursor whose own winner stayed, even + /// though the route the cursor cannot see left and so changed the prefix's + /// overall winner. + #[moq_net_sim::test] + async fn a_moved_path_restarts_a_scoped_cursor() { + let (producer, claim) = scoped_pool(); + let consumer = producer.consume(); + let _a = claim("room", "room/video/a", 10, 5); + let b = claim("room", "room/video/b", 11, 9); + let chat = claim("room", "room/chat", 12, 1); + let mut video = consumer + .clone() + .scope("", &scopes(&["room/video"])) + .unwrap() + .announced(); + assert_eq!(video.assert_next_active("room").cost, Cost::new(5)); + let _sticky = hold(&consumer, "room/video/b", &b).await; + + drop(chat); + let _d = claim("room", "room/video/b", 13, 8); + assert_eq!(restarted(&mut video, "room").await.cost, Cost::new(5)); + quiet(&mut video).await; + } + + /// A requester joining the front after its cursor restarted is not covered by + /// that restart: the path moving afterwards restarts the prefix again. + #[moq_net_sim::test] + async fn a_moved_path_restarts_a_requester_that_joined_after_a_restart() { + let (producer, claim) = scoped_pool(); + let consumer = producer.consume(); + let _w = claim("pool", "pool/other", 10, 2); + let r = claim("pool", "pool/job", 11, 5); + let mut announced = consumer.announced(); + assert_eq!(announced.assert_next_active("pool").cost, Cost::new(2)); + let (_answer, sticky) = hold(&consumer, "pool/job", &r).await; + + let _y = claim("pool", "pool/other", 12, 1); + assert_eq!(announced.assert_next_restarted("pool").cost, Cost::new(1)); + quiet(&mut announced).await; + // The path still resolves through the front's route, so the re-request joins it. + let joined = consumer.request_broadcast("pool/job", None).await.unwrap(); + assert!(joined.is_clone(&sticky)); + + let _z = claim("pool", "pool/job", 13, 3); + assert_eq!(restarted(&mut announced, "pool").await.cost, Cost::new(1)); + quiet(&mut announced).await; + } + + /// A path that moved with the prefix's own restart is settled: the prefix's winner + /// moving back to a route the front saw restarts it once, and the front adds none. + #[moq_net_sim::test] + async fn a_moved_path_settled_by_a_restart_stays_settled() { + let (producer, claim) = scoped_pool(); + let consumer = producer.consume(); + let a = claim("room", "room/job", 10, 5); + let _b = claim("room", "room/other", 11, 2); + let mut announced = consumer.announced(); + assert_eq!(announced.assert_next_active("room").cost, Cost::new(2)); + let _sticky = hold(&consumer, "room/job", &a).await; + + let c = producer + .dynamic("room", Route::default().with_hops(hops(&[12])).with_cost(1)) + .unwrap(); + assert_eq!(announced.assert_next_restarted("room").cost, Cost::new(1)); + quiet(&mut announced).await; + + c.update(Route::default().with_hops(hops(&[12])).with_cost(3)).unwrap(); + assert_eq!(announced.assert_next_restarted("room").cost, Cost::new(2)); + quiet(&mut announced).await; + } + #[test] fn exclude_hides_routes_through_the_peer() { let producer = origin(1).produce(); @@ -6564,6 +6701,7 @@ mod tests { horizon: Horizon::default(), watch, request: kio::Producer::default(), + joined: Arc::default(), timers: consumer.timers.clone(), }; let mut front = std::pin::pin!(serve_front(task)); From 80c0600da7ae23e898d2cbc29a810e9f12348dc6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 12:57:09 -0700 Subject: [PATCH 3/4] fix(net): count a cursor's switch to an older route as its restart A route that predates a request can still win the prefix and the held path after it, as in a cost failover between two existing routes. The cursor's switch to it is the restart, but its generation predates the join, so the front renewed the prefix again. Each cursor now stamps its presented route with next_route as it starts, and a join or a new front takes a value of its own, so the check compares when the cursor restarted rather than when the route was minted. Adds origin/pool_cursors, sweeping the paths a pool serves against the cursors watching it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/benches/origin.rs | 60 ++++++++++++++++++++++++++--- rs/moq-net/src/model/origin.rs | 70 ++++++++++++++++++++++++---------- 2 files changed, 104 insertions(+), 26 deletions(-) diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 82c75de8f5..8c55b21ef2 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -10,7 +10,7 @@ //! ranked once per change, then each cursor selects its first visible entry. //! //! An equal-cost pool is swept the same way, over its members and the paths it -//! already serves. +//! already serves, and over those paths and the cursors watching the pool. //! //! A route swap on one front is swept over its tracks and the copies each track //! still holds from earlier routes, and a front retiring over the fronts around it. @@ -465,23 +465,38 @@ fn bench_retire(c: &mut Criterion) { const POOL: [usize; 3] = [4, 32, 256]; const POOL_PATHS: [usize; 3] = [100, 1_000, 10_000]; +/// Announce cursors watching the pool, against [`POOL_PATHS`]. +const POOL_CURSORS: [usize; 3] = [1, 100, 1_000]; + /// Hop ids for pool members, clear of the origin's own. const POOL_HOP: u64 = 1_000; /// An origin where `members` equal-cost advertisers claim `pool`, already /// serving `paths` requested paths beneath it, spread across the pool by the -/// hash on each path. Every handle is held so the fronts stay up. +/// hash on each path, and watched by `cursors` announce cursors. Every handle +/// is held so the fronts stay up. struct Pool { producer: origin::Producer, driver: origin::Driver, _members: Vec, _producers: Vec, _consumers: Vec, + cursors: Vec, +} + +impl Pool { + /// Discard what the cursors were delivered, so each iteration measures only what it adds. + fn drain(&mut self) { + for cursor in &mut self.cursors { + while cursor.next().now_or_never().flatten().is_some() {} + } + } } -fn pool(members: usize, paths: usize) -> Pool { +fn pool(members: usize, paths: usize, cursors: usize) -> Pool { let (producer, mut driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap())); let consumer = producer.consume(); + let cursors = (0..cursors).map(|_| consumer.announced()).collect(); let members: Vec<_> = (0..members) .map(|i| producer.dynamic("pool", pool_route(POOL_HOP + i as u64)).unwrap()) .collect(); @@ -506,13 +521,16 @@ fn pool(members: usize, paths: usize) -> Pool { .map(|request| request.now_or_never().expect("resolved once driven").expect("served")) .collect(); - Pool { + let mut pool = Pool { producer, driver, _members: members, _producers: producers, _consumers: consumers, - } + cursors, + }; + pool.drain(); + pool } /// A pool member's claim: one hop, at the same cost as every other member. @@ -535,7 +553,35 @@ fn bench_pool_churn(c: &mut Criterion) { for paths in POOL_PATHS { let id = BenchmarkId::from_parameter(format!("{members}m_{paths}p")); group.bench_function(id, |b| { - let mut pool = pool(members, paths); + let mut pool = pool(members, paths, 0); + let waiter = kio::Waiter::noop(); + let hop = POOL_HOP + members as u64; + b.iter(|| { + let joined = pool.producer.dynamic("pool", pool_route(hop)).unwrap(); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + drop(joined); + pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + }); + }); + } + } + group.finish(); +} + +/// [`bench_pool_churn`] at 32 members, watched by `cursors` announce cursors. +/// +/// A front whose path the newcomer takes checks whether every cursor on the +/// prefix already restarted for it before renewing the prefix, so the moved +/// fronts times the cursors is the slope to watch. +fn bench_pool_cursors(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/pool_cursors"); + group.sample_size(10); + let members = 32; + for paths in POOL_PATHS { + for cursors in POOL_CURSORS { + let id = BenchmarkId::from_parameter(format!("{paths}p_{cursors}c")); + group.bench_function(id, |b| { + let mut pool = pool(members, paths, cursors); let waiter = kio::Waiter::noop(); let hop = POOL_HOP + members as u64; b.iter(|| { @@ -543,6 +589,7 @@ fn bench_pool_churn(c: &mut Criterion) { pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); drop(joined); pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); + pool.drain(); }); }); } @@ -881,6 +928,7 @@ criterion_group!( bench_request, bench_retire, bench_pool_churn, + bench_pool_cursors, bench_handoff, bench_relay, bench_parked, diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index c4a42df128..98715a3f2c 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -994,8 +994,8 @@ struct RemoteFront { } /// The last route a cursor observed: the instance it serves, its metadata, -/// servability, and captures. -type CursorRoute = (Instance, RouteMeta, bool, Option>); +/// servability, captures, and [`OriginState::next_route`] as it started. +type CursorRoute = (Instance, RouteMeta, bool, Option>, u64); impl WeakEntry for RemoteFront { fn is_closed(&self) -> bool { @@ -1086,22 +1086,29 @@ impl TableCursor { && self.named.as_ref().is_none_or(|(mount, _)| mount.names(&entry.prefix)) } - /// Deliver a changed winner at this presented prefix. - fn update(&mut self, presented: &PathOwned, best: Option<&RouteEntry>) { + /// Deliver a changed winner at this presented prefix, as of `now`, the + /// [`OriginState::next_route`]. + fn update(&mut self, presented: &PathOwned, best: Option<&RouteEntry>, now: u64) { match best { Some(entry) => { let meta = (entry.hops.clone(), entry.cost, entry.entered(), entry.epoch.clone()); let served = entry.server.is_some(); let captures = self.captures(&entry.prefix); let instance = entry.instance(); + let started = match self.current.get(presented) { + Some((prev, _, _, prev_captures, started)) if *prev == instance && *prev_captures == captures => { + *started + } + _ => now, + }; let previous = self.current.insert( presented.clone(), - (instance.clone(), meta.clone(), served, captures.clone()), + (instance.clone(), meta.clone(), served, captures.clone(), started), ); match previous { // Captures are consumer identity, not route metadata. Replace the old // identity explicitly so consumers keyed on it remove it. - Some((_, prev, _, prev_captures)) if prev_captures != captures => { + Some((_, prev, _, prev_captures, _)) if prev_captures != captures => { if let Ok(mut state) = self.state.write() { state.apply_unannounce(self.under.join(presented), prev, prev_captures); state.apply_announce(self.under.join(presented), meta, captures); @@ -1119,7 +1126,7 @@ impl TableCursor { // A servability flip is delivered: a request that failed Unroutable // under an advertise-only route retries on the update, and hiding // it would park that waiter forever. - Some((_, prev, prev_served, _)) if prev == meta && prev_served == served => {} + Some((_, prev, prev_served, ..)) if prev == meta && prev_served == served => {} _ => { if let Ok(mut state) = self.state.write() { state.apply_announce(self.under.join(presented), meta, captures); @@ -1128,7 +1135,7 @@ impl TableCursor { } } None => { - if let Some((_, last, _, captures)) = self.current.remove(presented) + if let Some((_, last, _, captures, _)) = self.current.remove(presented) && let Ok(mut state) = self.state.write() { state.apply_unannounce(self.under.join(presented), last, captures); @@ -3409,6 +3416,8 @@ struct OriginState { // The announced routes, keyed by prefix. The table holds one entry per live // advertisement, not one per broadcast consumer. routes: RouteTable, + // Route ids and generations, and the clock a cursor's start and a front's join + // read. A join takes a value of its own, so a start before it reads older. next_route: u64, next_watch: u64, @@ -3519,6 +3528,7 @@ impl OriginState { fn sync_route(&mut self, prefix: &Path, claim: &Pattern) { // Split borrows: the recompute reads `routes` while mutating a cursor. let routes = &self.routes; + let now = self.next_route; let mut candidates = None; for id in routes.cursors_touching(prefix) { let Some(cursor) = self.cursors.get_mut(&id) else { @@ -3527,7 +3537,7 @@ impl OriginState { if let Some(presented) = cursor.presented(prefix, claim) { if presented.is_empty() { // A cursor root can collapse several covering prefixes into one. - Self::sync_cursor(routes, cursor, &presented); + Self::sync_cursor(routes, cursor, &presented, now); } else { // Every non-root presentation refers to this exact prefix. Rank // its routes once, then take each cursor's first visible entry. @@ -3543,7 +3553,7 @@ impl OriginState { .iter() .map(|(_, entry)| *entry) .find(|entry| cursor.visible(entry)); - cursor.update(&presented, best); + cursor.update(&presented, best, now); } } } @@ -3583,8 +3593,8 @@ impl OriginState { } /// Recompute the best visible route presenting at `presented` (relative) for - /// one cursor and deliver the change, if any. - fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned) { + /// one cursor and deliver the change, if any, as of `now`. + fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned, now: u64) { // The entries presenting here are the ones announced at the absolute // prefix, or, for the cursor's own root, at the root and every prefix // above it (all of which present as the empty path). Among them, the @@ -3608,7 +3618,7 @@ impl OriginState { .min_by_key(|entry| route_order(&entry.prefix, entry)) }); - cursor.update(presented, best); + cursor.update(presented, best, now); } /// Register a cursor and replay the current best route per presented prefix. @@ -3629,7 +3639,7 @@ impl OriginState { } } for p in &presented { - Self::sync_cursor(&self.routes, &mut cursor, p); + Self::sync_cursor(&self.routes, &mut cursor, p, self.next_route); } for head in &cursor.heads { self.routes.add_cursor(head, id); @@ -3683,17 +3693,17 @@ impl OriginState { best } - /// Whether every cursor presenting `prefix` restarted it since route generation - /// `since`: none still presents an instance without an epoch from before, which - /// is what a [renewal](Self::renew) would restart. + /// Whether every cursor presenting `prefix` restarted it since + /// [`next_route`](Self::next_route) read `since`: none still presents an instance + /// without an epoch it started before, which is what a [renewal](Self::renew) + /// would restart. fn restarted_since(&self, prefix: &Path, since: u64) -> bool { - let old = |instance: &Instance| matches!(instance, Instance::Route(generation) if *generation < since); // Without a route there from before, no cursor can present one: the common // case once a renewal moved every generation, and cheaper than the cursors. let Some(claim) = self .routes .at(prefix) - .find(|entry| old(&entry.instance())) + .find(|entry| matches!(entry.instance(), Instance::Route(generation) if generation < since)) .map(|entry| &entry.claim) else { return true; @@ -3705,7 +3715,8 @@ impl OriginState { let current = cursor .presented(prefix, claim) .and_then(|presented| cursor.current.get(&presented)); - !current.is_some_and(|(instance, ..)| old(instance)) + // An older route the cursor switched to since is a restart all the same. + !current.is_some_and(|(instance, .., started)| matches!(instance, Instance::Route(_)) && *started < since) }) } } @@ -4700,6 +4711,7 @@ impl Consumer { .then(|| front.broadcast.consume()) .filter(|held| !held.is_closed()); if let Some(held) = held { + state.next_route += 1; front.joined.store(state.next_route, Ordering::Release); let pending = Requesting::queued(front.request.consume(), held) .with_path(requested) @@ -4725,6 +4737,7 @@ impl Consumer { // The front starts held by this request; see [`Front::new`]. let held = broadcast.consume(); let watch = state.watch(&self.shared, &absolute); + state.next_route += 1; let joined = Arc::new(AtomicU64::new(state.next_route)); state.fronts.insert( key, @@ -5967,6 +5980,23 @@ mod tests { quiet(&mut announced).await; } + /// A route from before the request repricing to win both the prefix and the held + /// path restarts the prefix once: the cursor's own switch covers the move. + #[moq_net_sim::test] + async fn a_repriced_route_taking_a_held_path_restarts_once() { + let (producer, claim) = scoped_pool(); + let consumer = producer.consume(); + let a = claim("room", "room", 10, 5); + let b = claim("room", "room", 11, 9); + let mut announced = consumer.announced(); + assert_eq!(announced.assert_next_active("room").cost, Cost::new(5)); + let _sticky = hold(&consumer, "room/job", &a).await; + + b.update(Route::default().with_hops(hops(&[11])).with_cost(3)).unwrap(); + assert_eq!(announced.assert_next_restarted("room").cost, Cost::new(3)); + quiet(&mut announced).await; + } + #[test] fn exclude_hides_routes_through_the_peer() { let producer = origin(1).produce(); From 526de635a0c563b40a9398389efb637e73ccdd0f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 13:16:55 -0700 Subject: [PATCH 4/4] fix(net): check a front's restart under the lock that renews The front decided under the read lock and renewed after releasing it, so a restart in between still left the renewal to restart the prefix again. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 98715a3f2c..1985b52973 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2500,7 +2500,11 @@ async fn serve_front(task: FrontTask) -> Vec { .and_then(|resolved| resolved.moved(&table, &path.as_path(), horizon, joined)); drop(table); if let Some(prefix) = renew { - shared.lock().renew(&prefix); + // Checked under the lock that renews: a restart after the read covers the move. + let mut table = shared.lock(); + if !table.restarted_since(&prefix, joined) { + table.renew(&prefix); + } } Event::Selected { best, serving_closing } }; @@ -3754,14 +3758,15 @@ struct Resolution { } impl Resolution { - /// The prefix to [renew](OriginState::renew): on a route without an epoch, a - /// request for the path, beneath the prefix, now resolves through another route - /// there. The prefix's own winner may be unchanged, so nothing else tells its - /// announce cursors that what they resolved under it moved. + /// The prefix to [renew](OriginState::renew) unless it already + /// [restarted](OriginState::restarted_since) since the last requester `joined` + /// the front: on a route without an epoch, a request for the path, beneath the + /// prefix, now resolves through another route there. The prefix's own winner may + /// be unchanged, so nothing else tells its announce cursors that what they + /// resolved under it moved. /// - /// A cursor that restarted since the last requester `joined` the front already - /// told everyone on it, and a renewal would restart it again. Settled once per - /// join: a renewal leaves the front's instance behind, so nobody joins it after. + /// Settled once per join: a renewal leaves the front's instance behind, so + /// nobody joins it after. fn moved(&mut self, table: &OriginState, path: &Path, horizon: Horizon, joined: u64) -> Option { if self.settled == Some(joined) || !matches!(self.instance, Instance::Route(_)) { return None; @@ -3776,10 +3781,7 @@ impl Resolution { return None; } self.settled = Some(joined); - match table.restarted_since(prefix, joined) { - true => None, - false => Some(prefix.clone()), - } + Some(prefix.clone()) } }