fix(p2p): collapse relay circuit dial fan-out and throttle denied dials

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<Multiaddr>) {
+        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

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::<ConnectError>(),
+            Some(ConnectError::ResourceLimitExceeded)
+        ) || matches!(
+            e.downcast_ref::<ReserveError>(),
+            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::<Vec<_>>()
-                    .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
+        ));
+    }
+}