backport: hold back a stale path to a prefix a cursor just retracted (#5375)

doc/bin/relay/cluster.md

@@ -27,7 +27,10 @@ So a change of a broadcast's best route waits 300 ms before it is announced,
 while a new broadcast and a removed one go out at once. By then the withdrawal
 has usually removed the other stale routes too, and the relay sends one
 retraction instead of advertising each stale path in turn. Requests still
-follow the current best route immediately; only the announcement waits.
+follow the current best route immediately; only the announcement waits. 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.
 
 Failover routes must carry copies of the same broadcast. For each track, the
 relay requires matching timescale, retention window, publisher priority, and

rs/moq-net/benches/origin.rs

@@ -12,7 +12,7 @@
 
 use std::time::Duration;
 
-use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
 use futures::FutureExt;
 use moq_net::{Hop, Hops, Pattern, Patterns, Timestamp, announce, broadcast, kio, origin};
 
@@ -81,6 +81,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 `<project>/.svc`. Every mount aliases the one target, the worst
@@ -480,6 +521,7 @@ fn bench_handoff(c: &mut Criterion) {
 criterion_group!(
 	benches,
 	bench_announce,
+	bench_withdraw,
 	bench_announce_mounted,
 	bench_announce_fleet,
 	bench_announce_duplicate,

rs/moq-net/src/model/origin.rs

@@ -137,13 +137,24 @@ 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) on a
+	/// strictly worse route waits. A republish ranks the same and a reconnect names
+	/// a fresh first hop, so neither waits. Zero disables this too.
 	pub update_hold: Duration,
 }
 
 /// The default [`Config::update_hold`]: longer than a withdrawal takes to cross a
 /// 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 {
@@ -651,13 +662,39 @@ fn route_order(prefix: &Path, entry: &RouteEntry) -> (bool, Cost, bool, usize, u
 /// The `(hops, cost, source)` metadata an announce cursor delivers alongside a prefix.
 type RouteMeta = (Hops, Cost, Source);
 
+/// The earlier of two optional deadlines.
+fn earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant> {
+	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 and rank, and when the quarantine ends.
+#[derive(Clone, Copy)]
+struct Quarantine {
+	held: Held,
+	publisher: Hop,
+	rank: (Cost, usize),
+}
+
+impl Quarantine {
+	/// Whether `meta` could be a stale path to the retracted publisher: the same
+	/// first hop, ranked strictly worse. A republish ranks the same and a reconnect
+	/// names a fresh first hop, so neither waits.
+	fn holds(&self, meta: &RouteMeta) -> bool {
+		meta.0.iter().next() == Some(&self.publisher) && (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
 /// 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),
@@ -734,19 +771,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<PathOwned, Held>,
+		retracted: &HashMap<PathOwned, Quarantine>,
 		now: Option<Instant>,
 		hold: Duration,
 	) -> Result<PathOwned, Option<Instant>> {
 		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() {
@@ -4501,6 +4552,11 @@ pub struct AnnounceConsumer {
 	// clock they count on, and the timer that wakes the next.
 	update_hold: Duration,
 	held: HashMap<PathOwned, Held>,
+	// 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<PathOwned, Quarantine>,
+	anchoring: VecDeque<(Instant, PathOwned)>,
+	expiring: VecDeque<(Instant, PathOwned)>,
 	timers: Clock,
 	hold: crate::runtime::Deadline<Clock>,
 }
@@ -4543,11 +4599,81 @@ impl AnnounceConsumer {
 			park: kio::Park::default(),
 			update_hold,
 			held: HashMap::new(),
+			retracted: HashMap::new(),
+			anchoring: VecDeque::new(),
+			expiring: VecDeque::new(),
 			hold: crate::runtime::Deadline::new(&timers),
 			timers,
 		}
 	}
 
+	/// Start a retraction's quarantine. A delivered announcement leaves it to
+	/// expire: later ones for the prefix are updates, which it never holds.
+	fn quarantine(&mut self, update: &AnnounceUpdate, now: Option<Instant>) {
+		if update.kind.is_active() {
+			return;
+		}
+		let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else {
+			return;
+		};
+		// 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 {
+			self.retracted.remove(&update.prefix);
+			return;
+		};
+		// A prefix retracted again at the same reading is already queued, so churn
+		// between clock readings can't grow the queue past one entry per prefix.
+		let queued = self
+			.retracted
+			.get(&update.prefix)
+			.is_some_and(|q| q.held == Held::Anchoring(now));
+		// The reading may be stale; [`Self::settle`] starts the quarantine on a fresh one.
+		let quarantine = Quarantine {
+			held: Held::Anchoring(now),
+			publisher,
+			rank: (update.route.cost, update.route.hops.len()),
+		};
+		self.retracted.insert(update.prefix.clone(), quarantine);
+		if !queued {
+			self.anchoring.push_back((now, update.prefix.clone()));
+		}
+	}
+
+	/// 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<Instant>) -> Option<Instant> {
+		let now = now?;
+		// 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;
+			}
+			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()));
+			}
+			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.
 	fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate {
 		let absolute = self.root.join(&update.prefix).to_owned();
@@ -4581,8 +4707,10 @@ impl AnnounceConsumer {
 	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<AnnounceUpdate>> {
 		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 update = match self.state.poll(waiter, |state| {
@@ -4592,7 +4720,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(())
@@ -4615,7 +4743,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;
 				}
@@ -4624,6 +4752,7 @@ impl AnnounceConsumer {
 				return Poll::Ready(None);
 			};
 			self.held.remove(&update.prefix);
+			self.quarantine(&update, now);
 			return Poll::Ready(Some(self.hand_out(update)));
 		}
 	}
@@ -4634,10 +4763,12 @@ impl AnnounceConsumer {
 	/// Use [`Self::is_closed`] to check if the cursor is closed.
 	pub fn try_next(&mut self) -> Option<AnnounceUpdate> {
 		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);
@@ -4648,6 +4779,7 @@ impl AnnounceConsumer {
 		self.held.remove(&prefix);
 		let update = state.take_prefix(prefix)?;
 		drop(state);
+		self.quarantine(&update, now);
 		Some(self.hand_out(update))
 	}
 
@@ -7083,6 +7215,59 @@ 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.
+	#[tokio::test(start_paused = true)]
+	async fn a_retraction_quarantines_a_named_publisher() {
+		let producer = origin(1).produce();
+		let peer = producer.clone().peer();
+		let mut announced = producer.consume().announced();
+		tokio::time::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 = tokio::time::Instant::now();
+		let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap();
+		let update = announced.next().await.unwrap();
+		assert_eq!(update.kind, AnnounceKind::Announced);
+		assert_eq!(update.route.hops, hops(&[9, 3, 4]));
+		assert!(start.elapsed() >= 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");
+
+		// Retract and announce cycles at one clock reading queue the prefix once.
+		for _ in 0..4 {
+			let churn = peer.dynamic("churn", route(&[9, 2], 2)).unwrap();
+			announced.assert_next_active("churn");
+			drop(churn);
+			announced.assert_next_ended("churn");
+		}
+		assert_eq!(announced.anchoring.len(), 1);
+
+		// A quarantine 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(), 2);
+		for _ in 0..2 {
+			announced.assert_next_wait();
+			tokio::time::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.