feat(relay): propagate errors to Transport::{listen_on,dial}

Cargo.toml

@@ -70,7 +70,7 @@ resolver = "2"
 rust-version = "1.73.0"
 
 [workspace.dependencies]
-futures-bounded = { version = "0.2.0", path = "misc/futures-bounded" }
+futures-bounded = { version = "0.2.1", path = "misc/futures-bounded" }
 libp2p = { version = "0.53.0", path = "libp2p" }
 libp2p-allow-block-list = { version = "0.3.0", path = "misc/allow-block-list" }
 libp2p-autonat = { version = "0.12.0", path = "protocols/autonat" }

misc/futures-bounded/CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.2.1 - unreleased
+
+- Add `.len()` getter to `FuturesMap`, `FuturesSet`, `StreamMap` and `StreamSet`.
+  See [PR XXXX](https://github.com/libp2p/rust-lib2pp/pulls/XXXX).
+
 ## 0.2.0
 
 - Add `StreamMap` type and remove `Future`-suffix from `PushError::ReplacedFuture` to reuse it for `StreamMap`.

misc/futures-bounded/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-bounded"
-version = "0.2.0"
+version = "0.2.1"
 edition = "2021"
 rust-version.workspace = true
 license = "MIT"

misc/futures-bounded/src/futures_map.rs

@@ -84,6 +84,10 @@ where
         }
     }
 
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
     }

misc/futures-bounded/src/futures_set.rs

@@ -42,6 +42,10 @@ impl<O> FuturesSet<O> {
         }
     }
 
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
     }

misc/futures-bounded/src/stream_map.rs

@@ -88,6 +88,10 @@ where
         Some(inner)
     }
 
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
     }
@@ -256,7 +260,7 @@ mod tests {
 
         assert!(poll.is_pending());
         assert_eq!(
-            streams.inner.len(),
+            streams.len(),
             0,
             "resources of cancelled streams are cleaned up properly"
         );

misc/futures-bounded/src/stream_set.rs

@@ -44,6 +44,10 @@ where
         }
     }
 
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
     }

protocols/relay/Cargo.toml

@@ -37,6 +37,7 @@ libp2p-plaintext = { workspace = true }
 libp2p-swarm = { workspace = true, features = ["macros", "async-std"] }
 libp2p-yamux = { workspace = true }
 quickcheck = { workspace = true }
+libp2p-swarm-test = { workspace = true }
 
 # Passing arguments to the docsrs builder in order to properly document cfg's.
 # More information: https://docs.rs/about/builds#cross-compiling

protocols/relay/src/lib.rs

@@ -47,15 +47,12 @@ pub mod inbound {
     pub mod hop {
         pub use crate::protocol::inbound_hop::FatalUpgradeError;
     }
-    pub mod stop {
-        pub use crate::protocol::inbound_stop::FatalUpgradeError;
-    }
 }
 
 /// Types related to the relay protocol outbound.
 pub mod outbound {
     pub mod hop {
-        pub use crate::protocol::outbound_hop::FatalUpgradeError;
+        pub use crate::protocol::outbound_hop::{ConnectError, ProtocolViolation, ReserveError};
     }
     pub mod stop {
         pub use crate::protocol::outbound_stop::FatalUpgradeError;

protocols/relay/src/priv_client.rs

@@ -25,7 +25,7 @@ pub(crate) mod transport;
 
 use crate::multiaddr_ext::MultiaddrExt;
 use crate::priv_client::handler::Handler;
-use crate::protocol::{self, inbound_stop, outbound_hop};
+use crate::protocol::{self, inbound_stop};
 use bytes::Bytes;
 use either::Either;
 use futures::channel::mpsc::Receiver;
@@ -40,8 +40,7 @@ use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, FromSwarm
 use libp2p_swarm::dial_opts::DialOpts;
 use libp2p_swarm::{
     dummy, ConnectionDenied, ConnectionHandler, ConnectionId, DialFailure, NetworkBehaviour,
-    NotifyHandler, Stream, StreamUpgradeError, THandler, THandlerInEvent, THandlerOutEvent,
-    ToSwarm,
+    NotifyHandler, Stream, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
 };
 use std::collections::{hash_map, HashMap, VecDeque};
 use std::io::{Error, ErrorKind, IoSlice};
@@ -60,32 +59,15 @@ pub enum Event {
         renewal: bool,
         limit: Option<protocol::Limit>,
     },
-    ReservationReqFailed {
-        relay_peer_id: PeerId,
-        /// Indicates whether the request replaces an existing reservation.
-        renewal: bool,
-        error: StreamUpgradeError<outbound_hop::ReservationFailedReason>,
-    },
     OutboundCircuitEstablished {
         relay_peer_id: PeerId,
         limit: Option<protocol::Limit>,
     },
-    OutboundCircuitReqFailed {
-        relay_peer_id: PeerId,
-        error: StreamUpgradeError<outbound_hop::CircuitFailedReason>,
-    },
     /// An inbound circuit has been established.
     InboundCircuitEstablished {
         src_peer_id: PeerId,
         limit: Option<protocol::Limit>,
     },
-    /// An inbound circuit request has been denied.
-    InboundCircuitReqDenied { src_peer_id: PeerId },
-    /// Denying an inbound circuit request failed.
-    InboundCircuitReqDenyFailed {
-        src_peer_id: PeerId,
-        error: inbound_stop::UpgradeError,
-    },
 }
 
 /// [`NetworkBehaviour`] implementation of the relay client
@@ -253,32 +235,15 @@ impl NetworkBehaviour for Behaviour {
                     limit,
                 }
             }
-            handler::Event::ReservationReqFailed { renewal, error } => {
-                Event::ReservationReqFailed {
-                    relay_peer_id: event_source,
-                    renewal,
-                    error,
-                }
-            }
             handler::Event::OutboundCircuitEstablished { limit } => {
                 Event::OutboundCircuitEstablished {
                     relay_peer_id: event_source,
                     limit,
                 }
             }
-            handler::Event::OutboundCircuitReqFailed { error } => Event::OutboundCircuitReqFailed {
-                relay_peer_id: event_source,
-                error,
-            },
             handler::Event::InboundCircuitEstablished { src_peer_id, limit } => {
                 Event::InboundCircuitEstablished { src_peer_id, limit }
             }
-            handler::Event::InboundCircuitReqDenied { src_peer_id } => {
-                Event::InboundCircuitReqDenied { src_peer_id }
-            }
-            handler::Event::InboundCircuitReqDenyFailed { src_peer_id, error } => {
-                Event::InboundCircuitReqDenyFailed { src_peer_id, error }
-            }
         };
 
         self.queued_actions.push_back(ToSwarm::GenerateEvent(event))
@@ -337,7 +302,7 @@ impl NetworkBehaviour for Behaviour {
                         peer_id: relay_peer_id,
                         handler: NotifyHandler::One(*connection_id),
                         event: Either::Left(handler::In::EstablishCircuit {
-                            send_back,
+                            to_dial: send_back,
                             dst_peer_id,
                         }),
                     },
@@ -351,7 +316,7 @@ impl NetworkBehaviour for Behaviour {
                         self.pending_handler_commands.insert(
                             connection_id,
                             handler::In::EstablishCircuit {
-                                send_back,
+                                to_dial: send_back,
                                 dst_peer_id,
                             },
                         );