From 2953b81e88ad3a4ff82d6250731d20e7553ad3be Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Tue, 8 Sep 2026 13:54:13 +0200 Subject: [PATCH] fix(p2p): collapse relay circuit dial fan-out and throttle denied dials Closes #564. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/p2p/src/relay/dial.rs | 77 ++++++++++++- crates/p2p/src/relay/event.rs | 156 ++++++++++++++++++++++++-- crates/p2p/src/relay/manager.rs | 101 ++++++++++++----- crates/p2p/src/relay/manager/tests.rs | 132 +++++++++++++++++++--- 4 files changed, 417 insertions(+), 49 deletions(-) diff --git a/crates/p2p/src/relay/dial.rs b/crates/p2p/src/relay/dial.rs index fe29d187..65f080c3 100644 --- a/crates/p2p/src/relay/dial.rs +++ b/crates/p2p/src/relay/dial.rs @@ -30,6 +30,17 @@ const RELAY_BACKOFF_MAX: Duration = Duration::from_secs(120); /// Jitter factor applied to backoff delays. Matches Charon's /// `DefaultConfig.Jitter`. const RELAY_BACKOFF_JITTER: f64 = 0.2; +/// Floor applied to the next delay after a relay explicitly denied a circuit +/// for exceeding a resource limit. +/// +/// A `ResourceLimitExceeded` denial means the relay's circuit quota or +/// circuit-source rate limiter is exhausted (rust-libp2p's defaults are 30 +/// circuits / 2 min per peer and 60 / min per IP). Retrying on the normal +/// backoff ladder — which starts at 1s — keeps the bucket empty and turns a +/// transient overload into a self-sustaining flood, so a denial parks the +/// campaign for at least this long regardless of how early in the ladder it +/// is. +const RELAY_DENIAL_BACKOFF_MIN: Duration = Duration::from_secs(30); /// State of an in-flight dial campaign, polled to produce a `ToSwarm::Dial` /// event each time its backoff elapses. @@ -59,6 +70,32 @@ impl RelayDialState { sleep: Box::pin(sleep_until(Instant::now())), } } + + /// Replaces the address set in place, preserving the backoff schedule. + /// + /// Re-routing is re-evaluated whenever any relay enters or leaves + /// `Reserved`, so a flapping relay would otherwise rebuild this state from + /// scratch on every transition and reset `retry_count` to zero — the + /// campaign would then dial at the 1s base delay forever. Keeping the + /// counter and the pending deadline means the ladder survives route + /// churn. + pub(super) fn set_addrs(&mut self, addrs: Vec) { + self.addrs = addrs; + } + + /// Parks the campaign after the relay denied a circuit for exceeding a + /// resource limit: advances the backoff ladder and re-arms the timer for + /// at least [`RELAY_DENIAL_BACKOFF_MIN`]. + pub(super) fn throttle_denied(&mut self) -> Duration { + let delay = backoff_delay(self.retry_count).max(RELAY_DENIAL_BACKOFF_MIN); + self.retry_count = self.retry_count.saturating_add(1); + let deadline = Instant::now() + .checked_add(delay) + .unwrap_or_else(Instant::now); + self.sleep.as_mut().reset(deadline); + + delay + } } impl Stream for RelayDialState { @@ -87,8 +124,9 @@ impl Stream for RelayDialState { } /// Returns true if both slices contain the same multiaddrs (order-independent). -/// Used to decide whether a routing refresh actually expanded the available -/// circuit paths to a peer — if it did, the dial state's backoff is reset. +/// Used to decide whether a routing refresh actually changed the available +/// circuit paths to a peer — if it did, the dial state's address set is +/// refreshed in place (see [`RelayDialState::set_addrs`]). pub(super) fn addr_sets_equal(a: &[Multiaddr], b: &[Multiaddr]) -> bool { if a.len() != b.len() { return false; @@ -152,6 +190,41 @@ mod tests { } } + #[tokio::test] + async fn throttle_denied_floors_the_next_delay_and_advances_the_ladder() { + let mut state = RelayDialState::new( + RelayDialType::ClusterPeer, + PeerId::random(), + vec!["/ip4/10.0.0.1/tcp/9000".parse().expect("valid multiaddr")], + ); + + // Early in the ladder the normal delay is 1s; a denial must not let + // the campaign back onto that schedule. + let delay = state.throttle_denied(); + assert!( + delay >= RELAY_DENIAL_BACKOFF_MIN, + "denial backoff {delay:?} below the floor {RELAY_DENIAL_BACKOFF_MIN:?}" + ); + assert_eq!(state.retry_count, 1); + } + + #[tokio::test] + async fn throttle_denied_keeps_the_ladder_when_it_already_exceeds_the_floor() { + let mut state = RelayDialState::new( + RelayDialType::ClusterPeer, + PeerId::random(), + vec!["/ip4/10.0.0.1/tcp/9000".parse().expect("valid multiaddr")], + ); + state.retry_count = 50; + + let delay = state.throttle_denied(); + assert!( + delay >= RELAY_BACKOFF_MAX.mul_f64(1.0 - RELAY_BACKOFF_JITTER), + "a late-ladder denial must not shrink the delay to the floor, got {delay:?}" + ); + assert_eq!(state.retry_count, 51); + } + #[test] fn backoff_delay_grows_then_plateaus() { // Averaging out jitter, retry=1 should be larger than base and diff --git a/crates/p2p/src/relay/event.rs b/crates/p2p/src/relay/event.rs index e5648ce8..5732002c 100644 --- a/crates/p2p/src/relay/event.rs +++ b/crates/p2p/src/relay/event.rs @@ -2,7 +2,11 @@ //! //! [`RelayManager`]: super::RelayManager -use libp2p::{PeerId, swarm::DialError}; +use libp2p::{ + PeerId, + relay::outbound::hop::{ConnectError, ReserveError}, + swarm::DialError, +}; /// Events emitted by [`RelayManager`] to the swarm. /// @@ -73,12 +77,69 @@ pub enum RelayDialError { /// Connection was denied by a behaviour or upgrade step. #[error("denied: {0}")] Denied(String), + /// The relay refused the circuit (or the reservation) because one of its + /// resource limits was exceeded — its circuit/reservation quota or one of + /// its rate limiters. + /// + /// Unlike the other variants this is *the relay throttling us*, so the + /// caller must slow down rather than retry on the normal ladder; see + /// `RelayDialState::throttle_denied`. + #[error("relay resource limit exceeded: {0}")] + ResourceLimitExceeded(String), /// All transport attempts failed; details preserved as `addr: err`, /// joined by `; `. #[error("transport: {0}")] Transport(String), } +impl RelayDialError { + /// Whether the relay denied us for exceeding one of its resource limits. + pub fn is_resource_limit_exceeded(&self) -> bool { + matches!(self, Self::ResourceLimitExceeded(_)) + } +} + +/// Renders an error together with its `source()` chain as `a: b: c`. +/// +/// A relay circuit failure reaches the swarm as +/// `TransportError::Other(io::Error)` wrapping several layers of transport +/// adapters, and every layer's `Display` drops its source — the top-level +/// message is a useless `"Failed to connect to destination."`. Walking the +/// chain is what makes the actual denial readable in logs. +fn error_chain(err: &(dyn std::error::Error + 'static)) -> String { + let mut parts = Vec::new(); + let mut current = Some(err); + while let Some(e) = current { + let msg = e.to_string(); + if !msg.is_empty() { + parts.push(msg); + } + current = e.source(); + } + + parts.join(": ") +} + +/// Whether `err` or any error in its `source()` chain is a relay +/// resource-limit denial (`RESOURCE_LIMIT_EXCEEDED` in the HOP response). +fn is_resource_limit_exceeded(err: &(dyn std::error::Error + 'static)) -> bool { + let mut current = Some(err); + while let Some(e) = current { + if matches!( + e.downcast_ref::(), + Some(ConnectError::ResourceLimitExceeded) + ) || matches!( + e.downcast_ref::(), + Some(ReserveError::ResourceLimitExceeded) + ) { + return true; + } + current = e.source(); + } + + false +} + impl From<&DialError> for RelayDialError { fn from(err: &DialError) -> Self { match err { @@ -87,14 +148,25 @@ impl From<&DialError> for RelayDialError { DialError::DialPeerConditionFalse(_) => Self::Skipped, DialError::Aborted => Self::Aborted, DialError::WrongPeerId { .. } => Self::WrongPeerId, - DialError::Denied { cause } => Self::Denied(cause.to_string()), - DialError::Transport(errors) => Self::Transport( - errors + DialError::Denied { cause } => { + if is_resource_limit_exceeded(cause) { + Self::ResourceLimitExceeded(error_chain(cause)) + } else { + Self::Denied(cause.to_string()) + } + } + DialError::Transport(errors) => { + let detail = errors .iter() - .map(|(addr, e)| format!("{addr}: {e}")) + .map(|(addr, e)| format!("{addr}: {}", error_chain(e))) .collect::>() - .join("; "), - ), + .join("; "); + if errors.iter().any(|(_, e)| is_resource_limit_exceeded(e)) { + Self::ResourceLimitExceeded(detail) + } else { + Self::Transport(detail) + } + } } } } @@ -108,3 +180,73 @@ pub enum RelayDialType { /// Dial a relay server directly. RelayServer, } + +#[cfg(test)] +mod tests { + use libp2p::core::transport::TransportError; + + use super::*; + + /// The transport error a relay circuit denial actually produces: the + /// `ConnectError` sits several `source()` levels below the boxed + /// `io::Error` the swarm surfaces, and every intermediate `Display` drops + /// its source. + fn boxed_relay_error(inner: libp2p::relay::client::transport::Error) -> DialError { + DialError::Transport(vec![( + "/ip4/10.0.0.1/tcp/9000".parse().expect("valid multiaddr"), + TransportError::Other(std::io::Error::other(inner)), + )]) + } + + #[test] + fn transport_resource_limit_denial_is_classified_and_detailed() { + let err = RelayDialError::from(&boxed_relay_error( + libp2p::relay::client::transport::Error::Connect(ConnectError::ResourceLimitExceeded), + )); + + assert!(err.is_resource_limit_exceeded()); + assert!( + err.to_string().contains("resource limit exceeded"), + "the denial must survive into the message, got {err}" + ); + } + + #[test] + fn transport_reservation_resource_limit_denial_is_classified() { + let err = RelayDialError::from(&boxed_relay_error( + libp2p::relay::client::transport::Error::Reservation( + ReserveError::ResourceLimitExceeded, + ), + )); + + assert!(err.is_resource_limit_exceeded()); + } + + #[test] + fn other_transport_failures_are_not_classified_as_denials() { + let err = RelayDialError::from(&boxed_relay_error( + libp2p::relay::client::transport::Error::Connect(ConnectError::NoReservation), + )); + + assert!(!err.is_resource_limit_exceeded()); + assert!(matches!(err, RelayDialError::Transport(_))); + // The chain walk is also what keeps the real cause visible: the + // outermost Display is a bare "Failed to connect to destination.". + assert!( + err.to_string().contains("Relay has no reservation"), + "source chain must be rendered, got {err}" + ); + } + + #[test] + fn non_transport_dial_errors_are_unchanged() { + assert!(matches!( + RelayDialError::from(&DialError::NoAddresses), + RelayDialError::NoAddresses + )); + assert!(matches!( + RelayDialError::from(&DialError::Aborted), + RelayDialError::Aborted + )); + } +} diff --git a/crates/p2p/src/relay/manager.rs b/crates/p2p/src/relay/manager.rs index da5c3018..412ba9a2 100644 --- a/crates/p2p/src/relay/manager.rs +++ b/crates/p2p/src/relay/manager.rs @@ -361,25 +361,39 @@ impl RelayManager { .collect() } - /// Builds circuit dial addresses for reaching `target` through every - /// currently reserved relay: - /// `/.../p2p//p2p-circuit/p2p/`. + /// Builds circuit dial addresses for reaching `target`: + /// `/.../p2p//p2p-circuit/p2p/`, exactly ONE per + /// currently reserved relay. + /// + /// A relay's transport addresses are alternative routes to the *same* + /// relay, not alternative routes to `target`. libp2p dials every address + /// in a `DialOpts` concurrently, so bundling one circuit address per + /// (relay × relay transport addr) fires N simultaneous HOP `CONNECT` + /// requests per peer pair — with three relay addresses a single campaign + /// burned three circuit-rate-limiter tokens within milliseconds and the + /// relay answered `ResourceLimitExceeded` for the rest of the run. + /// + /// Collapsing to one address per relay costs nothing: we only route + /// through *reserved* relays, so a transport connection to the relay is + /// already open, and rust-libp2p's circuit client reuses it + /// (`priv_client::Behaviour` handling of `DialReq`) — the transport + /// address embedded in the circuit multiaddr is never dialed. This + /// mirrors the reservation side, which likewise requests exactly one + /// listener per relay. fn peer_circuit_addrs(&self, target: &PeerId) -> Vec { let mut addrs = Vec::new(); for relay_id in self.reserved_relay_ids() { - let Some(relay_addrs) = self.relay_addrs.get(&relay_id) else { + let Some(relay_addr) = self.relay_addrs.get(&relay_id).and_then(|a| a.first()) else { continue; }; - for relay_addr in relay_addrs { - let mut circuit: Multiaddr = relay_addr - .iter() - .filter(|p| !matches!(p, MaProtocol::P2p(_))) - .collect(); - circuit.push(MaProtocol::P2p(relay_id)); - circuit.push(MaProtocol::P2pCircuit); - circuit.push(MaProtocol::P2p(*target)); - addrs.push(circuit); - } + let mut circuit: Multiaddr = relay_addr + .iter() + .filter(|p| !matches!(p, MaProtocol::P2p(_))) + .collect(); + circuit.push(MaProtocol::P2p(relay_id)); + circuit.push(MaProtocol::P2pCircuit); + circuit.push(MaProtocol::P2p(*target)); + addrs.push(circuit); } addrs } @@ -404,13 +418,17 @@ impl RelayManager { /// Inserts or refreshes a dial state for `target` using the current circuit /// addrs. /// - /// If the address set changed (or there was no dial state yet) the backoff - /// schedule is reset so the new route is tried immediately. If the address - /// set is unchanged, the existing dial state is left alone — its backoff - /// schedule survives so we don't hammer peers that have been unreachable - /// just because re-routing was re-evaluated. If no reserved relay can - /// currently reach `target`, any pre-existing dial state is removed so we - /// don't keep firing `Dial` events at circuits through unreserved relays. + /// An existing campaign keeps its backoff schedule: if the address set is + /// unchanged it is left alone entirely, and if it changed only the + /// addresses are swapped in place. Re-routing is re-evaluated on every + /// relay `Reserved` transition (and on every watchdog sweep), so + /// rebuilding the state would reset `retry_count` to zero and pin a + /// flapping route to the 1s base delay forever — the campaign would never + /// back off no matter how consistently the relay denies it. + /// + /// If no reserved relay can currently reach `target`, any pre-existing + /// dial state is removed so we don't keep firing `Dial` events at circuits + /// through unreserved relays. fn upsert_peer_dial(&mut self, target: PeerId) { let addrs = self.peer_circuit_addrs(&target); if addrs.is_empty() { @@ -418,9 +436,10 @@ impl RelayManager { return; } - if let Some(existing) = self.dial_states.get(&target) - && addr_sets_equal(&existing.addrs, &addrs) - { + if let Some(existing) = self.dial_states.get_mut(&target) { + if !addr_sets_equal(&existing.addrs, &addrs) { + existing.set_addrs(addrs); + } return; } @@ -655,6 +674,12 @@ impl RelayManager { /// backoff retries are cheap (libp2p re-rejects with the same error) and /// `on_connection_established` will tear the dial state down once libp2p /// surfaces the connection. + /// + /// The other special case is a relay `ResourceLimitExceeded` denial: the + /// relay is telling us its circuit quota or rate limiter is exhausted, so + /// the campaign is parked via `RelayDialState::throttle_denied` instead of + /// retrying on the normal ladder (which restarts at 1s and keeps the + /// bucket empty). fn on_dial_failure(&mut self, peer_id: Option, error: &DialError) { let Some(peer_id) = peer_id else { return }; let Some(state) = self.dial_states.get(&peer_id) else { @@ -663,6 +688,32 @@ impl RelayManager { let target = state.ty; let retry_count = state.retry_count; let skipped = matches!(error, DialError::DialPeerConditionFalse(_)); + let dial_error = RelayDialError::from(error); + + if dial_error.is_resource_limit_exceeded() { + // `state` is borrowed immutably above; re-borrow mutably here. + let backoff = self + .dial_states + .get_mut(&peer_id) + .map(RelayDialState::throttle_denied); + tracing::warn!( + peer_id = %peer_id, + dial_type = ?target, + retry_count, + ?backoff, + %dial_error, + "Relay denied circuit for exceeding a resource limit; throttling dial campaign" + ); + self.events + .push_back(ToSwarm::GenerateEvent(RelayManagerEvent::DialFailed { + peer_id, + target, + retry_count, + error: dial_error, + })); + + return; + } if skipped { match target { @@ -711,7 +762,7 @@ impl RelayManager { peer_id, target, retry_count, - error: RelayDialError::from(error), + error: dial_error, })); } diff --git a/crates/p2p/src/relay/manager/tests.rs b/crates/p2p/src/relay/manager/tests.rs index 1fa16daf..38f5b660 100644 --- a/crates/p2p/src/relay/manager/tests.rs +++ b/crates/p2p/src/relay/manager/tests.rs @@ -152,16 +152,17 @@ fn peer_circuit_addrs_skips_reserved_relay_without_tracked_addrs() { } #[test] -fn peer_circuit_addrs_builds_one_circuit_per_reserved_relay_addr() { +fn peer_circuit_addrs_builds_exactly_one_circuit_per_reserved_relay() { let mut mgr = manager(); let target = PeerId::random(); let relay = PeerId::random(); let relay_addrs = vec![ - // With and without trailing /p2p/ — both should produce the - // same canonical circuit form. + // With and without trailing /p2p/ — both would produce the + // same canonical circuit form, but only the first addr is used. addr(&format!("/ip4/10.0.0.1/tcp/9000/p2p/{relay}")), addr("/ip4/10.0.0.1/udp/9000/quic-v1"), + addr("/ip4/10.0.0.2/tcp/9000"), ]; mgr.connection_states .insert(relay, RelayConnectionState::Reserved); @@ -169,15 +170,16 @@ fn peer_circuit_addrs_builds_one_circuit_per_reserved_relay_addr() { let out = mgr.peer_circuit_addrs(&target); - let expected = vec![ - addr(&format!( + // libp2p dials every address in a DialOpts concurrently, so one circuit + // address per relay transport addr means N simultaneous HOP requests + // against the same relay. The relay's transport addrs are alternative + // routes to the relay, not to the target, so exactly one is emitted. + assert_eq!( + out, + vec![addr(&format!( "/ip4/10.0.0.1/tcp/9000/p2p/{relay}/p2p-circuit/p2p/{target}" - )), - addr(&format!( - "/ip4/10.0.0.1/udp/9000/quic-v1/p2p/{relay}/p2p-circuit/p2p/{target}" - )), - ]; - assert_eq!(out, expected); + ))] + ); } #[test] @@ -687,6 +689,90 @@ async fn on_dial_failure_skipped_relay_keeps_dial_state() { ); } +// ---- on_dial_failure: resource-limit denial ----------------------- + +/// A `DialError` shaped like a real relay circuit denial: the relay answered +/// the HOP `CONNECT` with `RESOURCE_LIMIT_EXCEEDED`, which reaches the swarm +/// as a boxed transport error several `source()` levels deep. +fn resource_limit_dial_error() -> DialError { + use libp2p::{ + core::transport::TransportError, + relay::{client::transport::Error as RelayTransportError, outbound::hop::ConnectError}, + }; + + DialError::Transport(vec![( + addr("/ip4/10.0.0.1/tcp/9000"), + TransportError::Other(std::io::Error::other(RelayTransportError::Connect( + ConnectError::ResourceLimitExceeded, + ))), + )]) +} + +#[tokio::test] +async fn on_dial_failure_resource_limit_throttles_campaign_and_reports_denial() { + let mut mgr = manager(); + let target = PeerId::random(); + let relay = PeerId::random(); + mgr.connection_states + .insert(relay, RelayConnectionState::Reserved); + mgr.relay_addrs + .insert(relay, vec![addr("/ip4/10.0.0.1/tcp/9000")]); + mgr.upsert_peer_dial(target); + + mgr.on_dial_failure(Some(target), &resource_limit_dial_error()); + + let state = mgr.dial_states.get(&target).expect("campaign stays armed"); + assert_eq!( + state.retry_count, 1, + "a denial must advance the backoff ladder, not leave it where it was" + ); + + let Some(ToSwarm::GenerateEvent(RelayManagerEvent::DialFailed { error, .. })) = + mgr.events.pop_front() + else { + panic!("expected a DialFailed event"); + }; + assert!( + error.is_resource_limit_exceeded(), + "denial must be classified as ResourceLimitExceeded, got {error}" + ); +} + +#[tokio::test] +async fn on_dial_failure_resource_limit_defers_next_dial_past_base_backoff() { + tokio::time::pause(); + + let mut mgr = manager(); + let target = PeerId::random(); + let relay = PeerId::random(); + mgr.connection_states + .insert(relay, RelayConnectionState::Reserved); + mgr.relay_addrs + .insert(relay, vec![addr("/ip4/10.0.0.1/tcp/9000")]); + mgr.upsert_peer_dial(target); + mgr.on_dial_failure(Some(target), &resource_limit_dial_error()); + + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + + // The normal ladder would already have fired again by now (base delay 1s); + // a denial parks the campaign for at least RELAY_DENIAL_BACKOFF_MIN. + tokio::time::advance(Duration::from_secs(10)).await; + mgr.events.clear(); + mgr.process_relay_dials(&mut cx); + assert!( + mgr.events.is_empty(), + "throttled campaign must not re-dial within the denial backoff window" + ); + + tokio::time::advance(Duration::from_secs(30)).await; + mgr.process_relay_dials(&mut cx); + assert!( + mgr.events.iter().any(|e| matches!(e, ToSwarm::Dial { .. })), + "campaign must resume once the denial backoff window elapses" + ); +} + // ---- upsert_peer_dial --------------------------------------------- #[tokio::test] @@ -716,7 +802,7 @@ async fn upsert_peer_dial_preserves_backoff_when_addrs_unchanged() { } #[tokio::test] -async fn upsert_peer_dial_resets_backoff_when_addrs_change() { +async fn upsert_peer_dial_refreshes_addrs_in_place_and_keeps_backoff() { let mut mgr = manager(); let target = PeerId::random(); let relay_a = PeerId::random(); @@ -738,10 +824,26 @@ async fn upsert_peer_dial_resets_backoff_when_addrs_change() { .insert(relay_b, vec![addr("/ip4/10.0.0.2/tcp/9000")]); mgr.upsert_peer_dial(target); + let state = mgr.dial_states.get(&target).expect("dial state"); assert_eq!( - mgr.dial_states.get(&target).map(|s| s.retry_count), - Some(0), - "addr-set changed: dial state (and backoff) must be replaced" + state.retry_count, 5, + "addr-set changed: the backoff ladder must survive route churn, or a \ + flapping relay pins the campaign to the base delay forever" + ); + let routed: HashSet = state.addrs.iter().cloned().collect(); + assert_eq!( + routed, + [ + addr(&format!( + "/ip4/10.0.0.1/tcp/9000/p2p/{relay_a}/p2p-circuit/p2p/{target}" + )), + addr(&format!( + "/ip4/10.0.0.2/tcp/9000/p2p/{relay_b}/p2p-circuit/p2p/{target}" + )), + ] + .into_iter() + .collect::>(), + "the new relay's circuit must be picked up" ); }