diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index d64da538e7..9ed7e60a78 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -22,12 +22,19 @@ relayed from what the peer just withdrew, rather than falling back to them one by one. During reconnect, another session from that peer can still advertise the broadcast; an old session's withdrawal does not invalidate that route. -A relay two hops from the publisher's still holds routes relayed through others. -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. +A relay two hops away hears only that its neighbor withdrew, and may still hold +paths relayed from the same announcement through neighbors the withdrawal has +not reached yet. So when the route a relay advertised to its cluster peers is +withdrawn and another takes over, the relay withdraws the broadcast from its +peers at once and advertises the replacement one second later, if the +replacement still stands; anything it hears for that broadcast in the meantime +waits too. A stale path is withdrawn within that second and never spreads, so an +ended broadcast is retracted once per link instead of hunting through the mesh. +New broadcasts, retractions, and a better route while the advertised one still +stands go out at once. The cost is a genuine failover: a peer left with no other path sees the broadcast return +about a second after a link fails. Requests and subscriptions follow the +current best route immediately; only the announcement waits. Clients are never +held down. Failover routes must carry copies of the same broadcast. For each track, the relay requires matching timescale, retention window, publisher priority, and diff --git a/quest/m0/README.md b/quest/m0/README.md index 5ec1c79c84..8263867ee7 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -47,7 +47,6 @@ Published API or wire breaks still land on dev; each quest's Plan says so. - [quest check everywhere](/quest/m0/quest-check-everywhere.md) - `quest check` guards `main`, `dev`, and the line branches on push and PR, not only PRs into `main` - [noq reassembly cap](/quest/m0/noq-reassembly-cap.md) - noq carries quinn's stream reassembly cap and the connection receive window is finite by default - [qmux reset race](/quest/m0/qmux-reset-race.md) - qmux handles RESET_STREAM under one lock instead of panicking -- [Path hunting](/quest/m0/path-hunting.md) - a withdrawn path is retracted about once per relay instead of hunting stale alternatives, with a hold-down on route updates and no wire change - [Shared fronts](/quest/m0/shared-fronts.md) - viewer sessions share a front, so fronts scale with peers, not viewers - [Frame alloc budget](/quest/m0/frame-alloc-budget.md) - frame buffers pre-allocate within a per-session budget and otherwise grow with bytes received - [Handshake deadline](/quest/m0/handshake-deadline.md) - an unfinished handshake or slow HTTP header times out diff --git a/quest/m0/path-hunting.md b/quest/m0/path-hunting.md deleted file mode 100644 index 7ca416b246..0000000000 --- a/quest/m0/path-hunting.md +++ /dev/null @@ -1,49 +0,0 @@ -# [M] A withdrawn path stops hunting - -## Goal - -When a broadcast ends, each relay in a lite-06 mesh retracts it about once, -instead of re-announcing stale alternatives. Today one withdrawal on live's -34-relay graph costs hundreds to over a thousand extra announce starts, and -on Kick dev an ended name drew 575 to 57,805 mesh announce starts (median -about 13,000) while live names drew one flood (moq-dev/moq.pro#2116). New -and removed routes still propagate at once, and no wire changes. - -## Plan - -Promoted from the cluster routing line to m0 on 2026-09-30: it is today's -partial-mesh hunting that #4399 left open, and it is hurting production. - -Decided (2026-09-30), lite-06, no wire change: - -- A route *update*, where a relay's advertised route for a path switches to a - different route, waits a short hold-down before it is re-advertised. A new - path (nothing advertised yet) and a removal (nothing left) go out at once. -- So when the best route is withdrawn and an alternative remains, the relay - retracts at once and re-announces the alternative only if it survives the - hold-down. Stale alternatives are withdrawn inside that window and never - spread. A genuine failover, where a link dies but the origin still - publishes, is delayed by the hold-down; pick the value from the repro, and - report the failover cost it adds. -- Check the interaction with #4399's withdrawal hiding and with - subscriptions already pinned to a route (fronts keep serving from a live - source; only advertisements are held). - -Test first: land moq.pro's in-process repro upstream (real lite-06 sessions -over the mock transport on live's 34-relay graph, mocked time), asserting -about one retraction per relay for a single withdrawal from a non-hub -publisher, plus a failover test where a link dies and the path comes back -through another route after the hold-down. It must fail on `main` today. - -Then moq.pro moves its pin and confirms about one retraction wave per ended -name on Kick dev after a deploy the user approves (moq.pro's own quest). - -If no hold-down closes it, the fallback is a per-origin seqno on cluster -links in the wip lite version; report before building it. - -Public API: none expected. Wire: none. - -## Related - -- [Cluster routing](/quest/m1/cluster-routing/README.md) - tiers shrink the core graph this hunts on -- [Cross-relay delivery under bursts](/quest/m1/cross-relay-bursts.md) - closed broadcasts announced for minutes, the same symptom diff --git a/quest/m1/cluster-routing/README.md b/quest/m1/cluster-routing/README.md index 386f8655e9..07ceb9e437 100644 --- a/quest/m1/cluster-routing/README.md +++ b/quest/m1/cluster-routing/README.md @@ -39,9 +39,9 @@ link-state and existence-split design planned earlier that day: path vector with hop lists stays loop-free, path hunting is confined to the cores, liveness flooding (the simulator's dominant link-state cost) is never paid, and failure detection sets the outage window either way. The parked - implementation is #4631. If stale paths remain on the tiered layout, the - fix is [Path hunting](/quest/m0/path-hunting.md), promoted to m0: a hold-down - on route updates, with no wire change. + implementation is #4631. Stale paths on any layout are held down: a relay + advertises a withdrawn route's replacement to its peers only after 1 s, + with no wire change. - The line lands on `main`: its children are additive. The two breaking changes left it for `dev` on their own: [Remove `--hop`](/quest/m1/hop-removal.md) and diff --git a/rs/moq-net/src/driver.rs b/rs/moq-net/src/driver.rs index 41779a81ae..22c677e4bb 100644 --- a/rs/moq-net/src/driver.rs +++ b/rs/moq-net/src/driver.rs @@ -62,7 +62,6 @@ impl Driver { /// Panics if `now` is earlier than the previous poll or construction time. pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.clock.advance(now); - self.clock.register_driver(waiter); match self.state.poll(waiter) { Poll::Ready(Ok(())) => Err(Error::Closed), Poll::Ready(Err(err)) => Err(err), diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 8b11b56424..5de4a889e8 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -5147,12 +5147,7 @@ mod tests { async fn a_repricing_is_a_request_update() { const VERSION: Version = Version::Draft19; - // Forward the update at once: the hold is not what this checks. - let origin = crate::origin::Config { - update_hold: Duration::ZERO, - ..crate::origin::Config::new(crate::Hop::new(1).unwrap()) - } - .produce(); + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let _cold = origin .announce("cam", crate::origin::Route::default().with_cost(4)) .unwrap(); @@ -5219,12 +5214,7 @@ mod tests { async fn a_publisher_change_is_a_request_update() { const VERSION: Version = Version::Draft19; - // Forward the update at once: the hold is not what this checks. - let origin = crate::origin::Config { - update_hold: Duration::ZERO, - ..crate::origin::Config::new(crate::Hop::new(1).unwrap()) - } - .produce(); + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let publisher_a = crate::Hops::try_from(vec![crate::Hop::new(7).unwrap()]).unwrap(); let publisher_b = crate::Hops::try_from(vec![crate::Hop::new(8).unwrap()]).unwrap(); let _from_a = origin diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index c967dc2750..f5f5b6ba98 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -40,8 +40,18 @@ pub(super) struct PublisherConfig { /// The origin (hop) id assigned to the peer, used whenever the peer doesn't /// declare one itself. See `Client::with_peer_hop`. pub peer_hop: Option, + /// Whether the peer is a cluster relay that re-advertises what we announce. + pub relay: bool, } +/// How long a peer relay waits for the replacement when the route we advertised +/// is withdrawn. A withdrawal must cross the mesh within this, so a replacement +/// derived from the withdrawn announcement is gone before it would spread: on +/// moq.pro's live graph that takes about 300 ms (`tests/mesh_withdraw.rs`), and +/// 1 s still holds at four times its estimated link latencies. It is also what a +/// genuine failover costs a peer left with no other path. +const HOLD_DOWN: Duration = Duration::from_secs(1); + /// Context shared by every control-stream child. struct Shared { session: S, @@ -56,6 +66,8 @@ struct Shared { // peer declines to declare one. Backs both the announce filter and the serving // origin, so a peer that names itself nowhere on the wire is still split-horizoned. peer_hop: Option, + // Whether the peer re-advertises what we announce: see `HOLD_DOWN`. + relay: bool, // The excluded origin handle, resolved once: the peer sends exactly one // SETUP, so its declared id never changes for the session. serving: std::sync::OnceLock, @@ -158,6 +170,7 @@ impl Publisher { self_origin, peer_setup: config.peer_setup, peer_hop: config.peer_hop, + relay: config.relay, serving: std::sync::OnceLock::new(), priority: Default::default(), version: config.version, @@ -285,9 +298,11 @@ impl Control { unreachable!() }; self.state = match kind { - lite::ControlType::Announce => { - ControlState::Announce(AnnounceServe::new(self.shared.clone(), stream)) - } + lite::ControlType::Announce => ControlState::Announce(AnnounceServe::new( + self.shared.clone(), + self.runtime.clone(), + stream, + )), lite::ControlType::Subscribe => { ControlState::Subscribe(SubscribeServe::new(self.shared.clone(), stream)) } @@ -479,6 +494,7 @@ impl ProbeServe { /// demand, and the origin change. struct AnnounceServe { shared: Arc>, + runtime: crate::time::Clock, stream: Option>, state: AnnounceState, } @@ -501,9 +517,10 @@ enum AnnounceState { } impl AnnounceServe { - fn new(shared: Arc>, stream: Stream) -> Self { + fn new(shared: Arc>, runtime: crate::time::Clock, stream: Stream) -> Self { Self { shared, + runtime, stream: Some(stream), state: AnnounceState::Decode, } @@ -597,7 +614,10 @@ impl AnnounceServe { false => origin, }; let announced = origin.announced(); - let run = AnnounceRun::new(self.shared.self_origin, self.shared.version); + let mut run = AnnounceRun::new(self.shared.self_origin, self.shared.version); + if self.shared.relay && lite::restart_supported(self.shared.version) { + run = run.with_hold(&self.runtime, HOLD_DOWN); + } self.state = AnnounceState::Run { origin, announced, run }; } } @@ -615,9 +635,65 @@ struct AnnounceRun { // The routes the peer currently holds, keyed by the suffix under the requested // prefix. live: HashMap, + // Replacements for withdrawn routes, advertised once they survive the hold-down. + // Never set for peers that do not re-advertise. + hold: Option, phase: AnnouncePhase, } +/// Replacement routes waiting out [`HOLD_DOWN`] before they reach a peer relay. +/// +/// When the route we advertised is withdrawn and an alternative takes over, the +/// alternative is usually derived from the same announcement, reaching us through +/// a peer the withdrawal has not crossed yet. Advertising it would send the peer +/// hunting through every such path; waiting lets the withdrawal catch up first. +struct Hold { + delay: Duration, + runtime: crate::time::Clock, + /// Fires when the front of `due` comes due. + deadline: crate::runtime::Deadline, + /// Held suffixes in the order they come due. An entry whose suffix was released + /// or held again since is skipped. + due: std::collections::VecDeque<(crate::time::Instant, crate::PathOwned)>, + /// The latest route per held suffix and when it comes due. A held suffix is not + /// in [`AnnounceRun::live`]: the peer holds nothing for it. + routes: HashMap, +} + +impl Hold { + /// Hold `route` for `suffix`, keeping the deadline of a suffix already held. + fn hold(&mut self, suffix: crate::PathOwned, hops: Hops, cost: crate::origin::Cost) { + let at = match self.routes.get(&suffix) { + Some((at, ..)) => *at, + None => { + let at = self.runtime.now() + self.delay; + self.due.push_back((at, suffix.clone())); + at + } + }; + self.routes.insert(suffix, (at, hops, cost)); + } + + /// The next held route whose hold-down has passed. + fn poll_due(&mut self, waiter: &kio::Waiter) -> Poll<(crate::PathOwned, Hops, crate::origin::Cost)> { + loop { + let Some((at, suffix)) = self.due.front() else { + self.deadline.set(None); + return Poll::Pending; + }; + if self.routes.get(suffix).is_none_or(|(held, ..)| held != at) { + self.due.pop_front(); + continue; + } + self.deadline.set(Some(*at)); + ready!(self.deadline.poll(waiter)); + let (_, suffix) = self.due.pop_front().expect("front checked above"); + let (_, hops, cost) = self.routes.remove(&suffix).expect("held checked above"); + return Poll::Ready((suffix, hops, cost)); + } + } +} + /// What the peer holds for one advertised suffix. struct Advertised { /// The announce id, on versions that assign them. @@ -643,10 +719,23 @@ impl AnnounceRun { version, encoder: lite::AnnounceEncoder::new(version), live: HashMap::new(), + hold: None, phase: AnnouncePhase::Init, } } + /// Hold a replacement for a withdrawn route `delay` before advertising it. + fn with_hold(mut self, runtime: &crate::time::Clock, delay: Duration) -> Self { + self.hold = Some(Hold { + delay, + runtime: runtime.clone(), + deadline: crate::runtime::Deadline::new(runtime), + due: Default::default(), + routes: HashMap::new(), + }); + self + } + /// The chain and cost to put on the wire for `route`, or `None` when it must /// not be forwarded. fn outgoing(&self, route: &crate::origin::Route, absolute: &crate::Path) -> Option<(Hops, crate::origin::Cost)> { @@ -826,7 +915,17 @@ impl AnnounceRun { if let Poll::Ready(res) = stream.reader.poll_closed(&mut cx) { return Poll::Ready(res); } + // Queued updates come first, so a held route that was withdrawn while the + // writer was blocked is dropped rather than advertised. Once the cursor is + // idle, a held route is the current one. let Poll::Ready(next) = announced.poll_next(waiter) else { + if let Some(hold) = &mut self.hold + && let Poll::Ready((suffix, hops, cost)) = hold.poll_due(waiter) + { + tracing::debug!(route = %origin.absolute(&suffix), "announce after hold-down"); + self.start(stream, suffix, hops, cost)?; + continue; + } return Poll::Pending; }; @@ -841,47 +940,72 @@ impl AnnounceRun { let absolute = origin.absolute(&update.prefix); let suffix = update.prefix; - if !update.kind.is_active() { + let outgoing = match update.kind.is_active() { + true => self.outgoing(&update.route, &absolute), + false => None, + }; + let Some((hops, cost)) = outgoing else { + // Retracted, or a chain that must not be forwarded (reflected, or + // full): drop any held replacement and retract what the peer holds. + if let Some(hold) = &mut self.hold { + hold.routes.remove(&suffix); + } self.retract(stream, suffix, &absolute)?; continue; + }; + + // A replacement already waiting out the hold-down takes the newest route. + if let Some(hold) = &mut self.hold + && hold.routes.contains_key(&suffix) + { + hold.hold(suffix, hops, cost); + continue; } - match self.outgoing(&update.route, &absolute) { - Some((hops, cost)) => match self.live.get_mut(&suffix) { - // The peer would decode what it already holds. - Some(advertised) if advertised.hops == hops && advertised.cost == cost => {} - // A metadata update on a live advertisement: restart it in - // place (lite-05 restarts via a duplicate ANNOUNCE). - Some(advertised) if lite::restart_supported(self.version) => { - tracing::debug!(route = %absolute, "reannounce"); - advertised.hops = hops.clone(); - advertised.cost = cost; - match advertised.id { - Some(id) => { - let hops = self.encoder.update(id, hops); - stream - .writer - .buffer(&lite::AnnounceBroadcast::Restart { id, hops, cost })? - } - // lite-05: a duplicate ANNOUNCE, which assigns no id. - None => stream.writer.buffer(&lite::AnnounceBroadcast::Active { - suffix: lite::PathRef::literal(suffix), - hops: lite::HopsRef::literal(hops), - cost, - })?, + match self.live.get_mut(&suffix) { + // The peer would decode what it already holds. + Some(advertised) if advertised.hops == hops && advertised.cost == cost => {} + // The route we advertised is gone and another took over: retract now, + // and advertise the replacement only if it survives the hold-down. + // A new chain from the same neighbor is that neighbor's own update, + // which it already held down if it had to. + Some(advertised) + if self.hold.is_some() + && advertised.hops != hops + && advertised.hops.as_slice().last() != hops.as_slice().last() + && !origin.carries(&suffix, &advertised.hops) => + { + self.retract(stream, suffix.clone(), &absolute)?; + self.hold.as_mut().expect("checked above").hold(suffix, hops, cost); + } + // A metadata update on a live advertisement: restart it in + // place (lite-05 restarts via a duplicate ANNOUNCE). + Some(advertised) if lite::restart_supported(self.version) => { + tracing::debug!(route = %absolute, "reannounce"); + advertised.hops = hops.clone(); + advertised.cost = cost; + match advertised.id { + Some(id) => { + let hops = self.encoder.update(id, hops); + stream + .writer + .buffer(&lite::AnnounceBroadcast::Restart { id, hops, cost })? } + // lite-05: a duplicate ANNOUNCE, which assigns no id. + None => stream.writer.buffer(&lite::AnnounceBroadcast::Active { + suffix: lite::PathRef::literal(suffix), + hops: lite::HopsRef::literal(hops), + cost, + })?, } - // Pre-restart versions have no way to update a live - // advertisement; the peer keeps the original chain. - Some(_) => {} - None => { - tracing::debug!(route = %absolute, "announce"); - self.start(stream, suffix, hops, cost)?; - } - }, - // The chain must not be forwarded (reflected, or full): retract - // whatever the peer holds. - None => self.retract(stream, suffix, &absolute)?, + } + // Pre-restart versions have no way to update a live + // advertisement; the peer keeps the original chain. + Some(_) => {} + None => { + tracing::debug!(route = %absolute, "announce"); + self.start(stream, suffix, hops, cost)?; + } } } } @@ -1857,7 +1981,6 @@ mod announce_test { h.announcement .update(crate::origin::Route::default().with_hops(pub_hops()).with_cost(3)) .unwrap(); - tokio::time::sleep(crate::origin::DEFAULT_UPDATE_HOLD).await; settle().await; match h.wire.take_announces().as_slice() { [lite::AnnounceBroadcast::Restart { id: 0, hops, cost }] => { @@ -1909,12 +2032,10 @@ mod announce_test { let mut h = harness().await; let route = |cost| crate::origin::Route::default().with_hops(pub_hops()).with_cost(cost); h.announcement.update(route(u64::MAX)).unwrap(); - tokio::time::sleep(crate::origin::DEFAULT_UPDATE_HOLD).await; settle().await; assert_eq!(h.wire.take_announces().len(), 1, "expected the clamped restart"); h.announcement.update(route(u64::MAX - 1)).unwrap(); - tokio::time::sleep(crate::origin::DEFAULT_UPDATE_HOLD).await; settle().await; h.assert_idle(); } @@ -1984,7 +2105,6 @@ mod announce_test { .with_cost(u64::MAX), ) .unwrap(); - tokio::time::sleep(crate::origin::DEFAULT_UPDATE_HOLD).await; settle().await; match h.wire.take_announces().as_slice() { [lite::AnnounceBroadcast::Restart { cost, .. }] => { @@ -1993,6 +2113,170 @@ mod announce_test { other => panic!("expected a clamped restart, got {other:?}"), } } + + fn chain(hops: &[u64]) -> Hops { + Hops::try_from(hops.iter().map(|id| Hop::new(*id).unwrap()).collect::>()).unwrap() + } + + fn route(hops: &[u64], cost: u64) -> crate::origin::Route { + crate::origin::Route::default().with_hops(chain(hops)).with_cost(cost) + } + + /// An announce loop to a peer relay, which holds replacements down, over an + /// origin where "cam" is reached through relay 5 and, at a higher cost, relay 6. + struct Held { + origin: origin::Producer, + best: Option, + fallback: Option, + wire: Wire, + /// Whether the peer accepts writes. + open: kio::Producer, + task: tokio::task::JoinHandle>, + } + + impl Held { + async fn new() -> Self { + let origin = Hop::new(1).unwrap().produce(); + let best = origin.announce("cam", route(&[9, 5], 1)).unwrap(); + let fallback = origin.announce("cam", route(&[9, 6], 2)).unwrap(); + + let log = Log::default(); + let writes = log.writes.clone(); + let consumer = origin.consume(); + let open = kio::Producer::new(true); + let mut stream = Stream:: { + writer: Writer::new(SinkSend::gated(log, open.consume()), VERSION), + reader: Reader::new(PendingRecv, VERSION), + }; + let task = tokio::spawn(async move { + let mut announced = consumer.announced(); + let mut run = + AnnounceRun::new(consumer.hop(), VERSION).with_hold(&crate::time::Clock::tokio(), HOLD_DOWN); + kio::wait(|waiter| run.poll(&mut stream, &consumer, &mut announced, waiter)).await + }); + settle().await; + + let mut wire = Wire { writes, cursor: 0 }; + assert_eq!(wire.take_ok().active, 1); + assert_eq!(wire.take_announces().len(), 1, "expected the initial announce"); + Self { + origin, + best: Some(best), + fallback: Some(fallback), + wire, + open, + task, + } + } + + /// Let the peer accept writes, or block it. + fn set_open(&self, open: bool) { + let Ok(mut gate) = self.open.write() else { + panic!("gate closed") + }; + *gate = open; + } + + fn assert_idle(&self) { + self.wire.assert_quiet(); + assert!(!self.task.is_finished(), "the announce loop ended unexpectedly"); + } + } + + /// When the advertised route is withdrawn, the peer loses it at once and gets + /// the replacement only after the hold-down. + #[tokio::test(start_paused = true)] + async fn withdrawn_route_is_replaced_after_the_hold_down() { + let mut h = Held::new().await; + h.best = None; + settle().await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::EndedId { id: 0 }] => {} + other => panic!("expected the retraction, got {other:?}"), + } + + tokio::time::sleep(HOLD_DOWN - Duration::from_millis(2)).await; + h.assert_idle(); + tokio::time::sleep(Duration::from_millis(2)).await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::Active { hops, cost, .. }] => { + assert_eq!(hops, &lite::HopsRef::literal(chain(&[9, 6]))); + assert_eq!(*cost, crate::origin::Cost::new(2)); + } + other => panic!("expected the replacement, got {other:?}"), + } + h.assert_idle(); + } + + /// A replacement withdrawn during the hold-down never reaches the peer: the + /// withdrawal costs one retraction, not one per stale path. + #[tokio::test(start_paused = true)] + async fn replacement_withdrawn_during_the_hold_down_is_never_sent() { + let mut h = Held::new().await; + h.best = None; + settle().await; + assert_eq!(h.wire.take_announces().len(), 1, "expected the retraction"); + + h.fallback = None; + tokio::time::sleep(HOLD_DOWN * 2).await; + h.assert_idle(); + + // A later announcement is a new route, sent at once. + let _again = h.origin.announce("cam", route(&[9, 7], 1)).unwrap(); + settle().await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::Active { .. }] => {} + other => panic!("expected the new route, got {other:?}"), + } + } + + /// A replacement withdrawn while the peer was blocked is dropped, not sent once + /// the hold-down passes: the withdrawal queued behind it is applied first. + #[tokio::test(start_paused = true)] + async fn replacement_withdrawn_while_blocked_is_never_sent() { + let mut h = Held::new().await; + h.set_open(false); + h.best = None; + settle().await; + h.fallback = None; + tokio::time::sleep(HOLD_DOWN * 2).await; + + h.set_open(true); + settle().await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::EndedId { id: 0 }] => {} + other => panic!("expected only the retraction, got {other:?}"), + } + h.assert_idle(); + } + + /// A better route while the advertised one still stands restarts at once: + /// nothing about it can be stale. + #[tokio::test(start_paused = true)] + async fn better_route_restarts_at_once() { + let mut h = Held::new().await; + let _better = h.origin.announce("cam", route(&[9, 4], 0)).unwrap(); + settle().await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::Restart { id: 0, .. }] => {} + other => panic!("expected a restart, got {other:?}"), + } + h.assert_idle(); + } + + /// A new chain from the neighbor we already route through is that neighbor's own + /// update, which it held down itself if it had to: restart at once. + #[tokio::test(start_paused = true)] + async fn same_neighbor_update_restarts_at_once() { + let mut h = Held::new().await; + h.best.as_ref().unwrap().update(route(&[8, 5], 1)).unwrap(); + settle().await; + match h.wire.take_announces().as_slice() { + [lite::AnnounceBroadcast::Restart { id: 0, .. }] => {} + other => panic!("expected a restart, got {other:?}"), + } + h.assert_idle(); + } } /// Buffer the per-frame timing prefix when the track advertises a timescale: @@ -3600,6 +3884,7 @@ mod tests { peer_setup, goaway, peer_hop: Some(assigned), + relay: false, }); let serving = kio::wait(|waiter| publisher.shared.poll_serving_origin(waiter)).await; @@ -3701,6 +3986,7 @@ mod tests { peer_setup: crate::lite::PeerSetup::default(), goaway, peer_hop: None, + relay: false, }); ProbeServe::new(publisher.shared.clone(), publisher.runtime.clone(), stream) } @@ -3848,6 +4134,7 @@ mod tests { peer_setup, goaway, peer_hop: None, + relay: false, }); let mut request = Vec::new(); diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index f02013ed5e..5ec414f949 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -138,6 +138,9 @@ where // An unset half gets an empty origin: an empty publish origin announces nothing // (and answers the peer's announce-interest with an empty set), and an empty // subscribe origin issues no ANNOUNCE_PLEASE. + // Whether the peer is a cluster relay: we mark what it announces as peer + // routes, and it re-advertises what we announce to it. + let relay = subscribe.as_ref().is_some_and(origin::Producer::is_peer); let publish = publish.unwrap_or_else(|| origin::Producer::empty(Hop::random()).consume()); let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random())); @@ -172,6 +175,7 @@ where peer_setup: peer_setup.clone(), goaway: goaway.clone(), peer_hop, + relay, }); let subscriber = Subscriber::new(SubscriberConfig { runtime: runtime.clone(), diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 78bbe59c92..39937261fd 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -41,7 +41,7 @@ pub use timed::Timed; /// Publishing broadcasts, announcing routes, and consuming both through an origin. pub mod origin { pub use super::origin_impl::{ - Config, Consumer, Cost, DEFAULT_UPDATE_HOLD, Driver, Dynamic, Producer, Request, Requesting, Route, Source, + Config, Consumer, Cost, Driver, Dynamic, Producer, Request, Requesting, Route, Source, }; } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 300a46640f..e8e6976f14 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -126,24 +126,8 @@ pub struct Config { /// edge. Defaults to [`track::DEFAULT_MAX_AGE`], and [`Self::cache_duration`] /// still caps it. pub default_max_age: Duration, - - /// How long an announce cursor holds a changed route before delivering it. - /// - /// A new route and a removed one are delivered at once; only an update (a - /// prefix whose best route changes while it stays reachable) waits, and a - /// newer change during the wait replaces it. When a publisher withdraws, a - /// relay that loses its best route but still holds routes derived from the - /// withdrawn one sends nothing while the withdrawal wave removes them, then one - /// 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. - 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); - impl Default for Config { /// A fresh random hop with no byte target and the default idle expiry. fn default() -> Self { @@ -153,7 +137,6 @@ impl Default for Config { pool, cache_duration: Duration::MAX, default_max_age: track::DEFAULT_MAX_AGE, - update_hold: DEFAULT_UPDATE_HOLD, } } } @@ -651,18 +634,6 @@ 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); -/// 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)] -enum Held { - Anchoring(Instant), - Until(Instant), -} - /// One coalesced update queued for an `AnnounceConsumer`. /// /// At most one entry exists per prefix, so a slow consumer's pending set is @@ -727,48 +698,9 @@ impl OriginConsumerState { } } - /// Whether the pending entry at `prefix` changes a route the consumer holds - /// to another route, rather than adding or removing one. - fn is_update(&self, prefix: &PathOwned, update: &PendingUpdate) -> bool { - matches!(update, PendingUpdate::Announce(_)) && self.delivered.contains(prefix) - } - - /// 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. - fn scan( - &self, - held: &mut 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 { - return Ok(prefix.clone()); - }; - // 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() { - None => { - held.insert(prefix.clone(), Held::Anchoring(now)); - now + Duration::from_nanos(1) - } - Some(Held::Anchoring(seen)) if now <= seen => seen + Duration::from_nanos(1), - Some(Held::Anchoring(_)) => { - held.insert(prefix.clone(), Held::Until(now + hold)); - now + hold - } - Some(Held::Until(until)) if until <= now => return Ok(prefix.clone()), - Some(Held::Until(until)) => until, - }; - wake = Some(wake.map_or(until, |w: Instant| w.min(until))); - } - Err(wake) - } - - /// Take the pending update at `prefix` for delivery. - fn take_prefix(&mut self, prefix: PathOwned) -> Option { + /// Take one update to deliver to the consumer, if any. + fn take(&mut self) -> Option { + let prefix = self.pending.keys().next()?.clone(); let ((meta, captures), kind) = match self.pending.remove(&prefix).unwrap() { PendingUpdate::Announce(meta) => { // The consumer has seen this prefix before, so it is a metadata update. @@ -1415,7 +1347,7 @@ impl Producer { pub fn new(config: Config) -> (Self, Driver) { let (tasks, set) = TaskSet::new(); let scope = OriginScope::default(); - let shared = kio::Shared::new(OriginState::new(config.update_hold)); + let shared = kio::Shared::::default(); let timers = Clock::default(); let pool = config.pool.clone(); let producer = Self { @@ -1463,6 +1395,11 @@ impl Producer { self } + /// Whether [`Self::peer`] marked this handle. + pub(crate) fn is_peer(&self) -> bool { + self.peer + } + /// This origin's construction config. pub fn config(&self) -> Config { Config { @@ -1470,7 +1407,6 @@ impl Producer { pool: self.pool.clone(), cache_duration: self.cache_duration, default_max_age: self.default_max_age, - update_hold: self.shared.lock().update_hold, } } @@ -1497,7 +1433,7 @@ impl Producer { hop, scope: OriginScope::empty(), root: PathOwned::default(), - shared: kio::Shared::new(OriginState::new(DEFAULT_UPDATE_HOLD)), + shared: kio::Shared::default(), pool: cache::Pool::default(), cache_duration: Duration::MAX, default_max_age: track::DEFAULT_MAX_AGE, @@ -2093,7 +2029,6 @@ impl Driver { /// has drained. pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.timers.advance(now); - self.timers.register_driver(waiter); let result = self.state.poll(waiter); let gc = self.pool.gc(now); if result.is_ready() { @@ -3215,11 +3150,8 @@ impl RouteTable { /// Carried in a [`kio::Shared`], so producers, consumers, and handlers work /// under one lock. Broadcasts published here are route table entries like the /// routes announced from elsewhere; this holds everything that serves a path. +#[derive(Default)] struct OriginState { - // See [`Config::update_hold`]. No `Default`: every origin takes it from its - // config, so no path silently disables the hold. - update_hold: Duration, - // The announced routes, keyed by prefix. The table holds one entry per live // advertisement, not one per broadcast consumer. routes: RouteTable, @@ -3250,19 +3182,6 @@ struct OriginState { } impl OriginState { - fn new(update_hold: Duration) -> Self { - Self { - update_hold, - routes: RouteTable::default(), - next_route: 0, - next_watch: 0, - cursors: HashMap::new(), - fronts: WeakCache::default(), - withdrawn: HashMap::new(), - closed: false, - } - } - /// Whether a peer in `hops` withdrew `prefix`. fn withdrawn_through(&self, prefix: &Path, hops: &Hops) -> bool { if self.withdrawn.is_empty() { @@ -3908,6 +3827,19 @@ impl Consumer { self } + /// Whether a live route this consumer sees still covers `path` along exactly `hops`. + pub(crate) fn carries(&self, path: impl AsPath, hops: &Hops) -> bool { + let named = self.root.join(path.as_path()).to_owned(); + let Some(absolute) = self.scope.resolve(&named) else { + return false; + }; + let state = self.shared.read(); + state + .routes + .covering(&absolute) + .any(|entry| entry.live() && self.horizon.admits(entry) && entry.hops == *hops) + } + /// A view of the routes that entered here: every route a handle marked /// [`Producer::peer`] announced is hidden from [`Self::announced`] and never /// resolved by [`Self::request_broadcast`]. @@ -4016,14 +3948,7 @@ impl Consumer { if let Some(mount) = self.scope.mount(&self.root) { // A root the mount resolves past the depth limit has nothing to announce. let Some(root) = mount.resolve(&self.root) else { - return AnnounceConsumer::new( - self.root.clone(), - Vec::new(), - state, - self.stats.clone(), - &self.shared, - self.timers.clone(), - ); + return AnnounceConsumer::new(self.root.clone(), Vec::new(), state, self.stats.clone(), &self.shared); }; let cursors = vec![cursor( root, @@ -4033,14 +3958,7 @@ impl Consumer { PathOwned::default(), Vec::new(), )]; - return AnnounceConsumer::new( - self.root.clone(), - cursors, - state, - self.stats.clone(), - &self.shared, - self.timers.clone(), - ); + return AnnounceConsumer::new(self.root.clone(), cursors, state, self.stats.clone(), &self.shared); } // Otherwise the consumer's own cursor, minus the mounted subtrees, plus one @@ -4084,14 +4002,7 @@ impl Consumer { holes, ), ); - AnnounceConsumer::new( - self.root.clone(), - cursors, - state, - self.stats.clone(), - &self.shared, - self.timers.clone(), - ) + AnnounceConsumer::new(self.root.clone(), cursors, state, self.stats.clone(), &self.shared) } /// Returns a cheap duplicate of this read handle. @@ -4394,13 +4305,6 @@ pub struct AnnounceConsumer { // Holds the waiter a `Stream` poll registered; disjoint from `state` so the // borrow never collides with the body's. park: kio::Park, - - // Held updates (see [`Config::update_hold`]): their holds by prefix, the - // clock they count on, and the timer that wakes the next. - update_hold: Duration, - held: HashMap, - timers: Clock, - hold: crate::runtime::Deadline, } impl AnnounceConsumer { @@ -4410,13 +4314,10 @@ impl AnnounceConsumer { state: kio::Producer, stats: stats::Session, shared: &kio::Shared, - timers: Clock, ) -> Self { let mut ids = Vec::with_capacity(cursors.len()); - let update_hold; { let mut table = shared.lock(); - update_hold = table.update_hold; if table.closed { // A cursor on a dead origin is born ended. if let Ok(mut state) = state.write() { @@ -4439,10 +4340,6 @@ impl AnnounceConsumer { stats, guards: HashMap::new(), park: kio::Park::default(), - update_hold, - held: HashMap::new(), - hold: crate::runtime::Deadline::new(&timers), - timers, } } @@ -4477,53 +4374,29 @@ impl AnnounceConsumer { /// cursor is closed, or `Poll::Pending` after registering `waiter` to be /// notified when the next update arrives. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - loop { - let now = self.timers.try_now(); - let hold = self.update_hold; - let held = &mut self.held; - let mut ready = None; - let mut wake = None; - let update = match self.state.poll(waiter, |state| { - if state.pending.is_empty() { - return match state.ended { - true => Poll::Ready(()), - false => Poll::Pending, - }; - } - match state.scan(held, now, hold) { - Ok(prefix) => { - ready = Some(prefix); - Poll::Ready(()) - } - Err(at) => { - wake = at; - Poll::Pending - } + let update = { + let mut state = match ready!(self.state.poll(waiter, |state| { + if state.pending.is_empty() && !state.ended { + Poll::Pending + } else { + Poll::Ready(()) } - }) { - Poll::Ready(Ok(mut state)) => match ready { - Some(prefix) => state.take_prefix(prefix), - None => { - // Ended by the origin's teardown, pending updates already - // drained; close the channel so every closure signal agrees. - state.close(); - None - } - }, + })) { + Ok(state) => state, // Closed: discard the Ref so its MutexGuard doesn't escape this call. - Poll::Ready(Err(_)) => None, - Poll::Pending => { - self.hold.set(wake); - ready!(self.hold.poll(waiter)); - continue; - } - }; - let Some(update) = update else { - return Poll::Ready(None); + Err(_) => return Poll::Ready(None), }; - self.held.remove(&update.prefix); - return Poll::Ready(Some(self.hand_out(update))); - } + match state.take() { + Some(update) => update, + None => { + // Ended by the origin's teardown, pending updates already + // drained; close the channel so every closure signal agrees. + state.close(); + return Poll::Ready(None); + } + } + }; + Poll::Ready(Some(self.hand_out(update))) } /// Returns the next update without blocking. @@ -4531,21 +4404,7 @@ impl AnnounceConsumer { /// Returns None if there is no update available; NOT because the cursor is closed. /// 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 mut state = self.state.write().ok()?; - let prefix = match state.scan(&mut self.held, now, self.update_hold) { - Ok(prefix) => prefix, - Err(wake) => { - // 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); - self.hold.set(wake); - return None; - } - }; - self.held.remove(&prefix); - let update = state.take_prefix(prefix)?; - drop(state); + let update = self.state.write().ok()?.take()?; Some(self.hand_out(update)) } @@ -6598,70 +6457,6 @@ mod tests { assert_eq!(announced.assert_next_active("room/x").source(), Source::Peer(origin(8))); } - /// A caller that only polls with `try_next` still gets a held update once its - /// hold passes. Nothing else wakes the idle driver, so the hold's own timer - /// must, or the origin's clock never reaches the deadline. - #[tokio::test(start_paused = true)] - async fn try_next_delivers_a_held_update() { - // Let paused time pass and every task woken by it run. - async fn step(by: Duration) { - tokio::time::advance(by).await; - for _ in 0..10 { - tokio::task::yield_now().await; - } - } - - let producer = origin(1).produce(); - let peer = producer.clone().peer(); - let mut announced = producer.consume().announced(); - step(Duration::from_millis(1)).await; - - let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via)); - let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap(); - assert!(announced.try_next().unwrap().kind.is_active()); - - let _far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap(); - drop(near); - assert!(announced.try_next().is_none(), "the update is held"); - step(Duration::from_millis(1)).await; - assert!(announced.try_next().is_none(), "the update is still held"); - step(DEFAULT_UPDATE_HOLD).await; - let update = announced.try_next().expect("the hold passed"); - assert_eq!(update.kind, AnnounceKind::Updated); - assert_eq!(update.route.hops, hops(&[9, 3, 4])); - } - - /// A new route and a removal are delivered at once. A change of the best route - /// waits out the update hold, and a newer change during the wait replaces it. - #[tokio::test(start_paused = true)] - async fn a_changed_route_waits_out_the_hold() { - let producer = origin(1).produce(); - let peer = producer.clone().peer(); - let mut announced = producer.consume().announced(); - // Holds count on the driver's clock, which starts at its first poll. - 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("room", route(&[9, 2], 2)).unwrap(); - announced.assert_next_active("room"); - - // The best route changes twice; the consumer sees only the last, once. - let far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap(); - let farther = peer.dynamic("room", route(&[9, 5, 6, 7], 7)).unwrap(); - let start = tokio::time::Instant::now(); - drop(near); - drop(far); - let update = announced.next().await.unwrap(); - assert_eq!(update.kind, AnnounceKind::Updated); - assert_eq!(update.route.hops, hops(&[9, 5, 6, 7])); - assert_eq!(start.elapsed(), DEFAULT_UPDATE_HOLD); - announced.assert_next_wait(); - - // The last route goes: retracted at once. - drop(farther); - announced.assert_next_ended("room"); - } - /// 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. diff --git a/rs/moq-net/src/time.rs b/rs/moq-net/src/time.rs index d73b688d4e..bef0ba27b4 100644 --- a/rs/moq-net/src/time.rs +++ b/rs/moq-net/src/time.rs @@ -66,9 +66,6 @@ struct State { now: Option, next: u64, timers: BTreeMap<(Instant, u64), Arc>>, - /// The driver's poll, woken when a timer armed outside it becomes the earliest, - /// so the driver's sleep never outlasts a deadline it did not see. - driver: kio::WaiterList, } impl Clock { @@ -103,21 +100,6 @@ impl Clock { } } - /// The latest supplied instant, or `None` before the first. - pub(crate) fn try_now(&self) -> Option { - let state = self.0.lock().unwrap(); - #[cfg(test)] - if state.automatic { - return Some(tokio::time::Instant::now().into_std()); - } - state.now - } - - /// Register the driver's poll to be woken by a new earliest timer. - pub(crate) fn register_driver(&self, waiter: &kio::Waiter) { - waiter.register(&mut self.0.lock().unwrap().driver); - } - pub(crate) fn timeout(&self) -> Option { self.0.lock().unwrap().timers.first_key_value().map(|((at, _), _)| *at) } @@ -178,15 +160,6 @@ impl crate::runtime::Timer for Timer { && state.now.is_none_or(|now| at > now) { state.timers.insert((at, self.id), self.waiters.clone()); - if state - .timers - .first_key_value() - .is_some_and(|(key, _)| *key == (at, self.id)) - { - let mut driver = state.driver.take(); - drop(state); - driver.wake(); - } } } fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { diff --git a/rs/moq-net/tests/mesh_withdraw.rs b/rs/moq-net/tests/mesh_withdraw.rs index 65d04646cb..4214ca88c3 100644 --- a/rs/moq-net/tests/mesh_withdraw.rs +++ b/rs/moq-net/tests/mesh_withdraw.rs @@ -9,7 +9,7 @@ mod support; use std::{collections::HashMap, time::Duration}; -use tokio::sync::mpsc; +use tokio::{sync::mpsc, time::Instant}; use moq_net::{Hop, Version, announce, broadcast, origin}; use support::harness::{MockPair, peer, peer_with_latency}; @@ -20,26 +20,40 @@ fn produce_origin(hop: u64) -> origin::Producer { producer } -/// Every update per prefix until the watcher goes quiet. Time is paused, so the -/// timeout fires only once every task is idle. -async fn drain( - watched: &mut mpsc::UnboundedReceiver<(String, announce::Kind)>, -) -> HashMap> { - let mut updates = HashMap::>::new(); - while let Ok(Some((prefix, kind))) = tokio::time::timeout(Duration::from_secs(1), watched.recv()).await { - updates.entry(prefix).or_default().push(kind); +/// One update a watcher saw: its prefix, kind, and when. +type Seen = (String, announce::Kind, Instant); + +/// Every update per prefix until the watcher goes quiet for longer than the +/// relays' hold-down. Time is paused, so the timeout fires only once every task +/// is idle and every held route has come due. +async fn drain_timed(watched: &mut mpsc::UnboundedReceiver) -> HashMap> { + let mut updates = HashMap::>::new(); + while let Ok(Some((prefix, kind, at))) = tokio::time::timeout(Duration::from_secs(5), watched.recv()).await { + updates.entry(prefix).or_default().push((kind, at)); } updates } +/// [`drain_timed`] without the timestamps. +async fn drain(watched: &mut mpsc::UnboundedReceiver) -> HashMap> { + drain_timed(watched) + .await + .into_iter() + .map(|(prefix, seen)| (prefix, seen.into_iter().map(|(kind, _)| kind).collect())) + .collect() +} + /// Watch `announced` from its own task, the way a session's announce writer does: /// it runs when woken, between the relays' own tasks, rather than only once the /// test task is polled again, which would coalesce every intermediate update. -fn watch(mut announced: announce::Consumer) -> mpsc::UnboundedReceiver<(String, announce::Kind)> { +fn watch(mut announced: announce::Consumer) -> mpsc::UnboundedReceiver { let (tx, rx) = mpsc::unbounded_channel(); tokio::spawn(async move { while let Some(update) = announced.next().await { - if tx.send((update.prefix.to_string(), update.kind)).is_err() { + if tx + .send((update.prefix.to_string(), update.kind, Instant::now())) + .is_err() + { break; } } @@ -51,7 +65,7 @@ fn watch(mut announced: announce::Consumer) -> mpsc::UnboundedReceiver<(String, struct Mesh { nodes: Vec, _pairs: Vec, - watched: mpsc::UnboundedReceiver<(String, announce::Kind)>, + watched: mpsc::UnboundedReceiver, } impl Mesh { @@ -145,76 +159,164 @@ async fn partial_mesh_withdraw_then_republish() { let _broadcasts = mesh.publish(100, 5).await; } -/// A 34-relay partial mesh: moq.pro's production graph on 2026-09-30, where -/// relays in neighboring PoPs peer and most reach a publisher's relay through -/// others. +/// moq.pro's live mesh on 2026-09-30: 34 relays, numbered by name from +/// `edge0.atl0` (0) to `edge2.sjc0` (33), peered with the relays in neighboring +/// PoPs. Each link carries its one-way latency in milliseconds, estimated from +/// the distance between its PoPs at 100 km/ms. #[rustfmt::skip] -const PRODUCTION: &[(usize, usize)] = &[ - (0, 20), (0, 22), (0, 29), (1, 3), (1, 12), (1, 14), (1, 21), (2, 20), (2, 26), (2, 32), (3, 8), - (3, 11), (3, 15), (3, 16), (3, 17), (3, 18), (3, 21), (3, 22), (3, 23), (3, 24), (3, 29), (3, 30), - (3, 31), (4, 7), (4, 20), (4, 22), (4, 25), (4, 29), (5, 20), (5, 27), (5, 33), (6, 23), (6, 24), - (6, 30), (6, 31), (7, 22), (7, 25), (7, 29), (8, 15), (8, 17), (8, 21), (8, 23), (8, 30), (9, 19), - (9, 22), (9, 23), (9, 28), (9, 29), (9, 30), (10, 19), (10, 20), (10, 22), (10, 23), (10, 26), - (10, 27), (10, 28), (10, 29), (10, 30), (10, 32), (10, 33), (11, 15), (11, 16), (11, 21), - (12, 14), (12, 27), (12, 33), (13, 14), (13, 27), (13, 33), (14, 26), (14, 27), (14, 32), - (14, 33), (15, 21), (16, 18), (16, 21), (17, 21), (17, 24), (17, 31), (18, 21), (18, 22), - (18, 23), (18, 24), (18, 29), (18, 30), (18, 31), (19, 22), (19, 23), (19, 26), (19, 28), - (19, 29), (19, 30), (19, 32), (20, 22), (20, 26), (20, 27), (20, 29), (20, 32), (20, 33), - (21, 22), (21, 23), (21, 24), (21, 29), (21, 30), (21, 31), (22, 23), (22, 24), (22, 25), - (22, 26), (22, 27), (22, 28), (22, 29), (22, 30), (22, 31), (22, 32), (22, 33), (23, 24), - (23, 28), (23, 29), (23, 30), (23, 31), (24, 29), (24, 30), (24, 31), (25, 29), (26, 27), - (26, 28), (26, 29), (26, 32), (26, 33), (27, 29), (27, 32), (27, 33), (28, 29), (28, 30), - (28, 32), (29, 30), (29, 31), (29, 32), (29, 33), (30, 31), (32, 33), +const LIVE: &[(usize, usize, u64)] = &[ + (0, 20, 12), (0, 22, 9), (0, 29, 9), (1, 3, 66), (1, 12, 39), (1, 14, 67), (1, 21, 66), + (2, 20, 11), (2, 26, 16), (2, 32, 16), (3, 8, 5), (3, 11, 4), (3, 15, 6), (3, 16, 9), + (3, 17, 3), (3, 18, 4), (3, 21, 1), (3, 22, 66), (3, 23, 6), (3, 24, 4), (3, 29, 66), + (3, 30, 6), (3, 31, 4), (4, 7, 66), (4, 20, 82), (4, 22, 76), (4, 25, 66), (4, 29, 76), + (5, 20, 20), (5, 27, 5), (5, 33, 5), (6, 23, 13), (6, 24, 13), (6, 30, 13), (6, 31, 13), + (7, 22, 15), (7, 25, 1), (7, 29, 15), (8, 15, 6), (8, 17, 2), (8, 21, 5), (8, 23, 10), + (8, 30, 10), (9, 19, 5), (9, 22, 4), (9, 23, 56), (9, 28, 5), (9, 29, 4), (9, 30, 56), + (10, 19, 12), (10, 20, 13), (10, 22, 9), (10, 23, 64), (10, 26, 28), (10, 27, 30), (10, 28, 12), + (10, 29, 9), (10, 30, 64), (10, 32, 28), (10, 33, 30), (11, 15, 3), (11, 16, 5), (11, 21, 4), + (12, 14, 53), (12, 27, 136), (12, 33, 136), (13, 14, 78), (13, 27, 120), (13, 33, 120), + (14, 26, 77), (14, 27, 83), (14, 32, 77), (14, 33, 83), (15, 21, 6), (16, 18, 11), (16, 21, 9), + (17, 21, 3), (17, 24, 5), (17, 31, 5), (18, 21, 4), (18, 22, 62), (18, 23, 4), (18, 24, 2), + (18, 29, 62), (18, 30, 4), (18, 31, 2), (19, 22, 8), (19, 23, 53), (19, 26, 37), (19, 28, 1), + (19, 29, 8), (19, 30, 53), (19, 32, 37), (20, 22, 19), (20, 26, 27), (20, 27, 24), (20, 29, 19), + (20, 32, 27), (20, 33, 24), (21, 22, 66), (21, 23, 6), (21, 24, 4), (21, 29, 66), (21, 30, 6), + (21, 31, 4), (22, 23, 59), (22, 24, 62), (22, 25, 15), (22, 26, 37), (22, 27, 39), (22, 28, 8), + (22, 29, 1), (22, 30, 59), (22, 31, 62), (22, 32, 37), (22, 33, 39), (23, 24, 2), (23, 28, 53), + (23, 29, 59), (23, 30, 1), (23, 31, 2), (24, 29, 62), (24, 30, 2), (24, 31, 1), (25, 29, 15), + (26, 27, 11), (26, 28, 37), (26, 29, 37), (26, 32, 1), (26, 33, 11), (27, 29, 39), (27, 32, 11), + (27, 33, 1), (28, 29, 8), (28, 30, 53), (28, 32, 37), (29, 30, 59), (29, 31, 62), (29, 32, 37), + (29, 33, 39), (30, 31, 2), (32, 33, 11), ]; -/// One-way latency in ms for each [`PRODUCTION`] link, from the great-circle -/// distance between its PoPs at 200 km/ms over a path 1.5 times as long, plus 1 ms. -#[rustfmt::skip] -const PRODUCTION_LATENCY_MS: &[u64] = &[ - 10, 7, 7, 50, 30, 51, 50, 9, 13, 13, 5, 4, 6, 8, 3, 4, 1, 50, 6, 4, 50, 6, 4, 50, 63, 58, 50, 58, - 16, 5, 5, 10, 10, 10, 10, 12, 1, 12, 6, 3, 5, 8, 8, 5, 4, 43, 5, 4, 43, 6, 11, 8, 49, 22, 23, 6, - 8, 49, 22, 23, 3, 5, 4, 41, 103, 103, 60, 91, 91, 59, 64, 59, 64, 6, 9, 8, 3, 6, 6, 4, 48, 4, 3, - 48, 4, 3, 5, 44, 26, 1, 5, 44, 26, 15, 21, 18, 15, 21, 18, 50, 6, 4, 50, 6, 4, 45, 47, 12, 29, 30, - 5, 1, 45, 47, 29, 30, 2, 44, 45, 1, 2, 47, 2, 1, 12, 9, 26, 29, 1, 9, 30, 9, 1, 5, 44, 26, 45, 47, - 29, 30, 2, 9, -]; +/// The live mesh over lite-06, every relay watched. +struct Live { + nodes: Vec, + /// One per [`LIVE`] link; `None` once the link failed. + links: Vec>, + watched: Vec>, +} -/// One withdrawal on a partial mesh retracts everywhere once. A relay two hops -/// from the publisher's hears only that its neighbor withdrew, and must not -/// fall back to, and re-advertise, another path derived from the same -/// announcement. Holding route updates (`origin::Config::update_hold`) lets the -/// withdrawal remove those paths first; with link latency and no hold, each -/// withdrawal here costs tens of thousands of announcements. -#[tokio::test(start_paused = true)] -async fn partial_mesh_withdraw_retracts_once() { - let version: Version = "moq-lite-06".parse().unwrap(); - let nodes: Vec<_> = (1..=34).map(produce_origin).collect(); - let mut pairs = Vec::new(); - for (&(a, b), &ms) in PRODUCTION.iter().zip(PRODUCTION_LATENCY_MS) { - pairs.push(peer_with_latency(version, &nodes[a], &nodes[b], Duration::from_millis(ms)).await); +impl Live { + /// `latency` scales each link's estimated latency: 0 for none, 1 for the estimate. + async fn new(latency: u32) -> Self { + let version: Version = "moq-lite-06".parse().unwrap(); + let nodes: Vec<_> = (1..=34).map(produce_origin).collect(); + let mut links = Vec::new(); + for &(a, b, ms) in LIVE { + let latency = Duration::from_millis(ms) * latency; + links.push(Some(peer_with_latency(version, &nodes[a], &nodes[b], latency).await)); + } + let watched = nodes.iter().map(|node| watch(node.consume().announced())).collect(); + Self { nodes, links, watched } } - let mut watched: Vec<_> = nodes.iter().map(|node| watch(node.consume().announced())).collect(); - for publisher in [5, 13, 22, 26, 32] { - let broadcast = nodes[publisher].create_broadcast(format!("room/{publisher}")).unwrap(); + /// Publish `name` from relay `publisher` and require every relay to see it. + async fn publish(&mut self, publisher: usize, name: &str) -> broadcast::Producer { + let broadcast = self.nodes[publisher].create_broadcast(name).unwrap(); broadcast.announce(Default::default()).unwrap(); - for (relay, watched) in watched.iter_mut().enumerate() { - let updates = drain(watched).await; + for (relay, updates) in self.drain().await.into_iter().enumerate() { + let kinds = updates.get(name).map(|seen| seen.last().unwrap().0); + assert!(kinds.is_some_and(|kind| kind.is_active()), "relay {relay}: {updates:?}"); + } + broadcast + } + + /// What each relay saw since the last drain. + async fn drain(&mut self) -> Vec>> { + let mut all = Vec::new(); + for watched in &mut self.watched { + all.push(drain_timed(watched).await); + } + all + } +} + +/// One withdrawal on the live mesh retracts everywhere once. A relay two hops +/// from the publisher's hears only that its neighbor withdrew, and switches to +/// another path derived from the same announcement. It must not advertise that +/// path before the withdrawal reaches it too, or its peers resurrect the name and +/// the mesh hunts through every stale path (moq-dev/moq.pro#2116). +async fn live_withdraw_retracts_once(latency: u32) { + let mut live = Live::new(latency).await; + for publisher in 0..live.nodes.len() { + let name = format!("room/{publisher}"); + let broadcast = live.publish(publisher, &name).await; + drop(broadcast); + for (relay, updates) in live.drain().await.into_iter().enumerate() { + // Stepping through the paths a relay already holds is local; announcing + // the name again, or retracting it more than once, is hunting. + let kinds: Vec<_> = updates[&name].iter().map(|(kind, _)| *kind).collect(); + let (last, stepped) = kinds.split_last().unwrap(); assert!( - updates.values().all(|kinds| kinds.last().unwrap().is_active()), - "relay {relay}: {updates:?}" + *last == announce::Kind::Retracted && stepped.iter().all(|kind| *kind == announce::Kind::Updated), + "publisher {publisher}, relay {relay}: {kinds:?}" ); } + } +} + +#[tokio::test(start_paused = true)] +async fn live_withdraw_retracts_once_without_latency() { + live_withdraw_retracts_once(0).await; +} + +#[tokio::test(start_paused = true)] +async fn live_withdraw_retracts_once_with_latency() { + live_withdraw_retracts_once(1).await; +} + +/// How long a relay holds a replacement before advertising it to a peer relay. +const HOLD_DOWN: Duration = Duration::from_secs(1); + +/// A link failure while the publisher stays live fails over: every relay ends up +/// with the broadcast again. A relay whose path crossed the dead link withdraws +/// it from its peers and re-advertises the replacement once it survives the +/// hold-down, so a relay left with no path at all waits about that long, and no +/// longer: the hold-downs along the mesh run side by side rather than in series. +#[tokio::test(start_paused = true)] +async fn live_link_failure_fails_over() { + let version: Version = "moq-lite-06".parse().unwrap(); + let mut live = Live::new(1).await; + let mut worst = Duration::ZERO; + for publisher in 0..live.nodes.len() { + let name = format!("room/{publisher}"); + let _broadcast = live.publish(publisher, &name).await; + + // Cut every link of the publisher's relay but its slowest. + let links: Vec<_> = (0..LIVE.len()) + .filter(|&index| LIVE[index].0 == publisher || LIVE[index].1 == publisher) + .collect(); + let kept = *links.iter().max_by_key(|&&index| LIVE[index].2).unwrap(); + let cut: Vec<_> = links.into_iter().filter(|&index| index != kept).collect(); + for &index in &cut { + live.links[index] = None; + } - drop(broadcast); - let mut hunted = Vec::new(); - for (relay, watched) in watched.iter_mut().enumerate() { - let updates = drain(watched).await; - let kinds = &updates[&format!("room/{publisher}")]; - if kinds != &[announce::Kind::Retracted] { - hunted.push((relay, kinds.clone())); + for (relay, updates) in live.drain().await.into_iter().enumerate() { + let Some(seen) = updates.get(&name) else { continue }; + assert!( + seen.last().unwrap().0.is_active(), + "publisher {publisher}, relay {relay}: {seen:?}" + ); + // Every stretch the relay spent with no path at all. + let mut gone = None; + for &(kind, at) in seen { + match (kind, gone.take()) { + (announce::Kind::Retracted, _) => gone = Some(at), + (_, Some(since)) => worst = worst.max(at - since), + _ => {} + } } } - assert!(hunted.is_empty(), "publisher {publisher}: {hunted:?}"); + + // Restore the links for the next publisher. + for &index in &cut { + let (a, b, ms) = LIVE[index]; + let latency = Duration::from_millis(ms); + live.links[index] = Some(peer_with_latency(version, &live.nodes[a], &live.nodes[b], latency).await); + } + live.drain().await; } + println!("longest a relay went without a path: {worst:?}"); + assert!(worst < 2 * HOLD_DOWN, "a relay went {worst:?} without a path"); }