From 58677d004b19232031969544bc1408cf957ff97a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 13:32:36 -0700 Subject: [PATCH 1/2] fix(net): hold back a stale path to a prefix a cursor just retracted Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/cluster.md | 4 +- rs/moq-net/src/model/origin.rs | 131 ++++++++++++++++++++++++++++-- rs/moq-net/tests/mesh_withdraw.rs | 37 +++++++++ 3 files changed, 165 insertions(+), 7 deletions(-) diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index 55f2b888957..e7c1cf058c0 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -24,7 +24,9 @@ unasked. moxygen announces nothing unasked on draft 14. When a moq-lite-04 or later peer withdraws a broadcast, the routes relayed through it go with it at once. A change of best route is announced after 300 ms, so stale paths are retracted once instead of advertised in turn; -requests follow the new best route immediately. +requests follow the new best route immediately. For 2 s after a relay retracts +a broadcast, it holds back a worse path to the same publisher that a peer still +hunting offers, instead of announcing the broadcast again. Routes carrying the same [publisher epoch](/concept/moq-lite#publisher-epochs) serve one broadcast. When the route serving it dies, withdraws, or is beaten by diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index f750d74b4b4..5d17cf7ccef 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -125,6 +125,12 @@ pub struct Config { /// retraction, instead of advertising each stale path in turn. It must outlast /// the wave's spread across the mesh. Defaults to [`DEFAULT_UPDATE_HOLD`]; /// zero delivers updates at once. + /// + /// A relay further out can still pass a stale path to a cursor that already + /// retracted the prefix, which would announce it anew. So for two seconds after + /// a retraction, an announcement from the same publisher (first hop and epoch) + /// on a strictly worse route waits. A republish ranks the same and a new + /// instance carries a new epoch, so neither waits. Zero disables this too. pub update_hold: Duration, } @@ -132,6 +138,11 @@ pub struct Config { /// global mesh, with margin. pub const DEFAULT_UPDATE_HOLD: Duration = Duration::from_millis(300); +/// How long a cursor holds back a stale path to a prefix it retracted; see +/// [`Config::update_hold`]. Path hunting after a withdrawal settles within it on +/// random meshes with links of up to 250 ms. +const RETRACT_QUARANTINE: Duration = Duration::from_secs(2); + impl Default for Config { /// A fresh random hop with no byte target and the default idle expiry. fn default() -> Self { @@ -606,6 +617,33 @@ fn route_order<'a>( /// The `(hops, cost, source, epoch)` metadata an announce cursor delivers alongside a prefix. type RouteMeta = (Hops, Cost, Source, Option); +/// The earlier of two optional deadlines. +fn earliest(a: Option, b: Option) -> Option { + match (a, b) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + } +} + +/// A retraction's quarantine, keyed by prefix in its [`AnnounceConsumer`]: the +/// retracted route's publisher, epoch, and rank, and when the quarantine ends. +#[derive(Clone)] +struct Quarantine { + held: Held, + publisher: Option, + epoch: Option, + rank: (Cost, usize), +} + +impl Quarantine { + /// Whether `meta` could be a stale path to the retracted publisher: the same + /// first hop and epoch, ranked strictly worse. A republish ranks the same and a + /// new instance carries a new epoch, so neither waits. + fn holds(&self, meta: &RouteMeta) -> bool { + meta.0.iter().next().copied() == self.publisher && meta.3 == self.epoch && (meta.1, meta.0.len()) > self.rank + } +} + /// A pending update's hold, keyed by prefix in its [`AnnounceConsumer`]. /// /// A consumer reads the origin's clock outside the driver, where it holds the @@ -716,19 +754,33 @@ impl OriginConsumerState { } /// The first pending prefix ready to deliver, or when the next held one is. - /// See [`Config::update_hold`]; a hold starts the first time a scan sees it. + /// See [`Config::update_hold`]; a hold starts the first time a scan sees it, + /// and a quarantine in `retracted` (settled against `now`) holds until it ends. fn scan( &self, held: &mut HashMap, + retracted: &HashMap, now: Option, hold: Duration, ) -> Result> { let mut wake = None; for (prefix, update) in &self.pending { - let holds = !hold.is_zero() && !self.ended && self.is_update(prefix, update); - let Some(now) = now.filter(|_| holds) else { + let Some(now) = now.filter(|_| !hold.is_zero() && !self.ended) else { return Ok(prefix.clone()); }; + if !self.is_update(prefix, update) { + let quarantine = match update { + PendingUpdate::Announce((meta, _)) => retracted.get(prefix).filter(|q| q.holds(meta)), + _ => None, + }; + let until = match quarantine.map(|q| q.held) { + Some(Held::Anchoring(seen)) => seen + Duration::from_nanos(1), + Some(Held::Until(until)) if now < until => until, + _ => return Ok(prefix.clone()), + }; + wake = Some(wake.map_or(until, |w: Instant| w.min(until))); + continue; + } // The clock reads the driver's last advance, which may be stale: anchor the // hold on the first advance after the update arrived, by waking the driver. let until = match held.get(prefix).copied() { @@ -5020,6 +5072,9 @@ pub struct AnnounceConsumer { // clock they count on, and the timer that wakes the next. update_hold: Duration, held: HashMap, + retracted: HashMap, + anchoring: Vec, + retracted_swept: usize, timers: Clock, hold: crate::time::Deadline, } @@ -5062,11 +5117,69 @@ impl AnnounceConsumer { park: kio::Park::default(), update_hold, held: HashMap::new(), + retracted: HashMap::new(), + anchoring: Vec::new(), + retracted_swept: 16, hold: crate::time::Deadline::new(&timers), timers, } } + /// Start a retraction's quarantine, or end it with a delivered announcement. + fn quarantine(&mut self, event: &AnnounceEvent, now: Option) { + let update = match event { + AnnounceEvent::End(announce) => announce, + AnnounceEvent::Start(announce) | AnnounceEvent::Update(announce) | AnnounceEvent::Restart(announce) => { + self.retracted.remove(&announce.prefix); + return; + } + }; + let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else { + return; + }; + // Forget expired quarantines once the map doubles, so the sweep is amortized. + if self.retracted.len() >= self.retracted_swept * 2 { + self.retracted + .retain(|_, q| !matches!(q.held, Held::Until(until) if until <= now)); + self.retracted_swept = self.retracted.len().max(16); + } + // The reading may be stale; [`Self::settle`] starts the quarantine on a fresh one. + let quarantine = Quarantine { + held: Held::Anchoring(now), + publisher: update.route.hops.iter().next().copied(), + epoch: update.route.epoch.clone(), + rank: (update.route.cost, update.route.hops.len()), + }; + self.retracted.insert(update.prefix.clone(), quarantine); + self.anchoring.push(update.prefix.clone()); + } + + /// Start the quarantines of retractions the clock has since moved past. Returns + /// when to wake to read the clock again, if any still wait. + fn settle(&mut self, now: Option) -> Option { + let now = now?; + let mut wake = None; + let retracted = &mut self.retracted; + self.anchoring.retain(|prefix| match retracted.get_mut(prefix) { + Some(Quarantine { + held: Held::Anchoring(seen), + .. + }) if now <= *seen => { + let at = *seen + Duration::from_nanos(1); + wake = Some(wake.map_or(at, |w: Instant| w.min(at))); + true + } + Some(Quarantine { held, .. }) => { + if matches!(held, Held::Anchoring(_)) { + *held = Held::Until(now + RETRACT_QUARANTINE); + } + false + } + None => false, + }); + wake + } + /// Drive the egress announce guards for one update. fn hand_out(&mut self, event: AnnounceEvent) -> AnnounceEvent { match &event { @@ -5102,8 +5215,10 @@ impl AnnounceConsumer { pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { loop { let now = self.timers.try_now(); + let settling = self.settle(now); let hold = self.update_hold; let held = &mut self.held; + let retracted = &self.retracted; let mut ready = None; let mut wake = None; let event = match self.state.poll(waiter, |state| { @@ -5113,7 +5228,7 @@ impl AnnounceConsumer { false => Poll::Pending, }; } - match state.scan(held, now, hold) { + match state.scan(held, retracted, now, hold) { Ok(prefix) => { ready = Some(prefix); Poll::Ready(()) @@ -5136,7 +5251,7 @@ impl AnnounceConsumer { // Closed: discard the Ref so its MutexGuard doesn't escape this call. Poll::Ready(Err(_)) => None, Poll::Pending => { - self.hold.set(wake); + self.hold.set(earliest(wake, settling)); ready!(self.hold.poll(waiter)); continue; } @@ -5149,6 +5264,7 @@ impl AnnounceConsumer { | AnnounceEvent::Restart(announce) | AnnounceEvent::End(announce)) = &event; self.held.remove(&announce.prefix); + self.quarantine(&event, now); return Poll::Ready(Some(self.hand_out(event))); } } @@ -5159,10 +5275,12 @@ impl AnnounceConsumer { /// Use [`Self::is_closed`] to check if the cursor is closed. pub fn try_next(&mut self) -> Option { let now = self.timers.try_now(); + let settling = self.settle(now); let mut state = self.state.write().ok()?; - let prefix = match state.scan(&mut self.held, now, self.update_hold) { + let prefix = match state.scan(&mut self.held, &self.retracted, now, self.update_hold) { Ok(prefix) => prefix, Err(wake) => { + let wake = earliest(wake, settling); // Arm the hold even though nothing polls it: the armed timer wakes the // driver, so its clock reaches the deadline for a later call to see. drop(state); @@ -5173,6 +5291,7 @@ impl AnnounceConsumer { self.held.remove(&prefix); let event = state.take_prefix(prefix); drop(state); + self.quarantine(&event, now); Some(self.hand_out(event)) } diff --git a/rs/moq-net/tests/mesh_withdraw.rs b/rs/moq-net/tests/mesh_withdraw.rs index b39b024b6a1..639c7f109ef 100644 --- a/rs/moq-net/tests/mesh_withdraw.rs +++ b/rs/moq-net/tests/mesh_withdraw.rs @@ -231,3 +231,40 @@ async fn partial_mesh_withdraw_retracts_once() { assert!(hunted.is_empty(), "publisher {publisher}: {hunted:?}"); } } + +/// A relay that already retracted a withdrawn broadcast must not announce it +/// again. Its neighbors, two or more hops from the publisher's relay, still hunt +/// through stale paths; one that falls back to a path not through this relay +/// offers it as a new route after the relay forgot the withdrawal. +#[moq_net_sim::test] +async fn retracted_relay_ignores_a_stale_path() { + #[rustfmt::skip] + const EDGES: &[((usize, usize), u64)] = &[ + ((0, 1), 35), ((1, 2), 90), ((0, 3), 129), ((2, 4), 140), ((1, 5), 137), ((4, 6), 148), + ((2, 7), 30), ((3, 5), 124), ((1, 4), 111), ((0, 5), 21), ((3, 4), 34), + ]; + let version: Version = "moq-lite-06".parse().unwrap(); + let nodes: Vec<_> = (1..=8).map(produce_origin).collect(); + let mut pairs = Vec::new(); + for &((a, b), ms) in EDGES { + pairs.push(peer_with_latency(version, &nodes[a], &nodes[b], Duration::from_millis(ms)).await); + } + let mut watched: Vec<_> = nodes.iter().map(|node| watch(node.consume().announced())).collect(); + + let broadcast = nodes[7].create_broadcast("room/7").unwrap(); + broadcast.announce(Default::default()).unwrap(); + for watched in watched.iter_mut() { + drain(watched).await; + } + + drop(broadcast); + let mut revived = Vec::new(); + for (relay, watched) in watched.iter_mut().enumerate() { + let kinds = drain(watched).await.remove("room/7").unwrap(); + assert_eq!(kinds.last(), Some(&Kind::End), "relay {relay}: {kinds:?}"); + if kinds.windows(2).any(|w| w == [Kind::End, Kind::Start]) { + revived.push((relay, kinds)); + } + } + assert!(revived.is_empty(), "{revived:?}"); +} From 2920eaa56a665073224ca471ecce31c7188842a8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 18:27:29 -0700 Subject: [PATCH 2/2] fix(net): settle retraction quarantines in clock order Anchoring rescanned every unanchored retraction on each poll, so draining N retractions read at one clock reading took O(N^2). Queue retractions in clock order and pop only the due fronts, which also forgets ended quarantines without waiting for the map to double. An anonymous route names no publisher, so it no longer starts a quarantine. Adds origin/withdraw: 10000 retractions drain in 22 ms, down from 1.04 s. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/benches/origin.rs | 42 +++++++++++ rs/moq-net/src/model/origin.rs | 124 +++++++++++++++++++++++---------- 2 files changed, 131 insertions(+), 35 deletions(-) diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 25fa255fc9d..88263158a8b 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -103,6 +103,47 @@ fn bench_announce(c: &mut Criterion) { group.finish(); } +/// A peer carrying `publishers` broadcasts withdraws them all before the +/// driver's clock moves again, then `subscribers` cursors drain the retractions. +/// Each retraction quarantines its prefix until a fresh clock reading, so a cost +/// per earlier retraction shows up as a slope over publishers. +fn bench_withdraw(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/withdraw"); + for (publishers, subscribers) in [(100, 1), (1_000, 1), (10_000, 1), (1_000, 10)] { + let id = BenchmarkId::from_parameter(format!("{publishers}p_{subscribers}s")); + group.bench_function(id, |b| { + b.iter_batched( + || { + let (producer, mut driver) = origin::Producer::new(origin::Config::default()); + // Retractions read the driver's last clock reading, which stays put below. + driver + .poll(moq_net::time::Instant::now(), &kio::Waiter::noop()) + .unwrap(); + let routes: Vec<_> = (0..publishers) + .map(|i| producer.dynamic(format!("room/{i}"), peer_route(9, 1)).unwrap()) + .collect(); + let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect(); + for cursor in &mut cursors { + while cursor.next().now_or_never().flatten().is_some() {} + } + (producer, driver, routes, cursors) + }, + |(producer, driver, routes, mut cursors)| { + drop(routes); + for cursor in &mut cursors { + for _ in 0..publishers { + cursor.next().now_or_never().flatten().expect("retract delivered"); + } + } + (producer, driver, cursors) + }, + BatchSize::LargeInput, + ); + }); + } + group.finish(); +} + /// `bench_announce` read through mounts: `publishers` routes under the fleet-wide /// `.svc/p0`, watched by `subscribers` project sessions that each mount it at /// their own `/.svc`. Every mount aliases the one target, the worst @@ -1078,6 +1119,7 @@ fn bench_copy_walk(c: &mut Criterion) { criterion_group!( benches, bench_announce, + bench_withdraw, bench_announce_mounted, bench_announce_fleet, bench_announce_duplicate, diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 5d17cf7ccef..aa9e9abb432 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -630,7 +630,7 @@ fn earliest(a: Option, b: Option) -> Option { #[derive(Clone)] struct Quarantine { held: Held, - publisher: Option, + publisher: Hop, epoch: Option, rank: (Cost, usize), } @@ -640,7 +640,7 @@ impl Quarantine { /// first hop and epoch, ranked strictly worse. A republish ranks the same and a /// new instance carries a new epoch, so neither waits. fn holds(&self, meta: &RouteMeta) -> bool { - meta.0.iter().next().copied() == self.publisher && meta.3 == self.epoch && (meta.1, meta.0.len()) > self.rank + meta.0.iter().next() == Some(&self.publisher) && meta.3 == self.epoch && (meta.1, meta.0.len()) > self.rank } } @@ -650,7 +650,7 @@ impl Quarantine { /// driver's last reading, not the time the update arrived. So a hold first waits /// for the driver's next reading (armed a nanosecond past the stale one, which /// wakes the driver at once), and counts from there. -#[derive(Clone, Copy)] +#[derive(Clone, Copy, PartialEq)] enum Held { Anchoring(Instant), Until(Instant), @@ -5072,9 +5072,11 @@ pub struct AnnounceConsumer { // clock they count on, and the timer that wakes the next. update_hold: Duration, held: HashMap, + // Quarantines by prefix, and their retractions in clock order: first waiting + // for a fresh reading (by the stale one), then quarantined (by when each ends). retracted: HashMap, - anchoring: Vec, - retracted_swept: usize, + anchoring: VecDeque<(Instant, PathOwned)>, + expiring: VecDeque<(Instant, PathOwned)>, timers: Clock, hold: crate::time::Deadline, } @@ -5118,8 +5120,8 @@ impl AnnounceConsumer { update_hold, held: HashMap::new(), retracted: HashMap::new(), - anchoring: Vec::new(), - retracted_swept: 16, + anchoring: VecDeque::new(), + expiring: VecDeque::new(), hold: crate::time::Deadline::new(&timers), timers, } @@ -5137,47 +5139,54 @@ impl AnnounceConsumer { let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else { return; }; - // Forget expired quarantines once the map doubles, so the sweep is amortized. - if self.retracted.len() >= self.retracted_swept * 2 { - self.retracted - .retain(|_, q| !matches!(q.held, Held::Until(until) if until <= now)); - self.retracted_swept = self.retracted.len().max(16); - } + // An anonymous route names no publisher to tie a later stale path to. + let Some(&publisher) = update.route.hops.iter().next().filter(|hop| **hop != Hop::UNKNOWN) else { + return; + }; // The reading may be stale; [`Self::settle`] starts the quarantine on a fresh one. let quarantine = Quarantine { held: Held::Anchoring(now), - publisher: update.route.hops.iter().next().copied(), + publisher, epoch: update.route.epoch.clone(), rank: (update.route.cost, update.route.hops.len()), }; self.retracted.insert(update.prefix.clone(), quarantine); - self.anchoring.push(update.prefix.clone()); + self.anchoring.push_back((now, update.prefix.clone())); } - /// Start the quarantines of retractions the clock has since moved past. Returns - /// when to wake to read the clock again, if any still wait. + /// Start the quarantines of retractions the clock has since moved past, and + /// forget ended ones. Returns when to wake to read the clock again, if any remain. fn settle(&mut self, now: Option) -> Option { let now = now?; - let mut wake = None; - let retracted = &mut self.retracted; - self.anchoring.retain(|prefix| match retracted.get_mut(prefix) { - Some(Quarantine { - held: Held::Anchoring(seen), - .. - }) if now <= *seen => { - let at = *seen + Duration::from_nanos(1); - wake = Some(wake.map_or(at, |w: Instant| w.min(at))); - true + // Both queues are in clock order, so only their fronts can be due. An entry + // whose prefix was since announced or retracted again no longer matches. + while let Some((seen, prefix)) = self.anchoring.front() { + if *seen >= now { + break; } - Some(Quarantine { held, .. }) => { - if matches!(held, Held::Anchoring(_)) { - *held = Held::Until(now + RETRACT_QUARANTINE); - } - false + if let Some(quarantine) = self.retracted.get_mut(prefix) + && quarantine.held == Held::Anchoring(*seen) + { + quarantine.held = Held::Until(now + RETRACT_QUARANTINE); + self.expiring.push_back((now + RETRACT_QUARANTINE, prefix.clone())); } - None => false, - }); - wake + self.anchoring.pop_front(); + } + while let Some((until, prefix)) = self.expiring.front() { + if *until > now { + break; + } + if self + .retracted + .get(prefix) + .is_some_and(|q| q.held == Held::Until(*until)) + { + self.retracted.remove(prefix); + } + self.expiring.pop_front(); + } + let anchoring = self.anchoring.front().map(|(seen, _)| *seen + Duration::from_nanos(1)); + earliest(anchoring, self.expiring.front().map(|(until, _)| *until)) } /// Drive the egress announce guards for one update. @@ -7967,6 +7976,51 @@ mod tests { announced.assert_next_ended("room"); } + /// After a retraction, a worse path to the same publisher waits out the + /// quarantine. An anonymous route names no publisher, so one after it is not held. + #[moq_net_sim::test] + async fn a_retraction_quarantines_a_named_publisher() { + let producer = origin(1).produce(); + let peer = producer.clone().peer(); + let mut announced = producer.consume().announced(); + moq_net_sim::sleep(Duration::from_millis(1)).await; + + let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via)); + + let near = peer.dynamic("named", route(&[9, 2], 2)).unwrap(); + announced.assert_next_active("named"); + drop(near); + announced.assert_next_ended("named"); + let start = moq_net_sim::now(); + let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap(); + match announced.next().await.unwrap() { + AnnounceEvent::Start(announce) => assert_eq!(announce.route.hops, hops(&[9, 3, 4])), + other => panic!("should be a start: got {other:?}"), + } + assert!(moq_net_sim::now() - start >= RETRACT_QUARANTINE); + + let near = peer.dynamic("anonymous", route(&[0, 2], 2)).unwrap(); + announced.assert_next_active("anonymous"); + drop(near); + announced.assert_next_ended("anonymous"); + let _other = peer.dynamic("anonymous", route(&[0, 3, 4], 4)).unwrap(); + announced.assert_next_active("anonymous"); + + // A quarantine nothing announces again is forgotten once it ends: the cursor + // wakes to anchor it, then again to drop it. + let gone = peer.dynamic("gone", route(&[9, 2], 2)).unwrap(); + announced.assert_next_active("gone"); + drop(gone); + announced.assert_next_ended("gone"); + assert_eq!(announced.retracted.len(), 1); + for _ in 0..2 { + announced.assert_next_wait(); + moq_net_sim::sleep(RETRACT_QUARANTINE).await; + } + announced.assert_next_wait(); + assert!(announced.retracted.is_empty(), "the ended quarantine is still held"); + } + /// A peer withdrawing a prefix hides the routes there through it, so the next /// best is never a path derived from the one withdrawn. Announcing again /// revives them, and nothing is remembered once no route passes through it.