diff --git a/Cargo.lock b/Cargo.lock index 020de1e3bc2..d97c3877359 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,6 +61,12 @@ dependencies = [ "cc", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "anes" version = "0.1.6" @@ -326,6 +332,15 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -1074,6 +1089,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-queue" +version = "0.3.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -1318,6 +1342,12 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "dotenvy" +version = "0.15.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" + [[package]] name = "dtls" version = "0.17.1" @@ -1410,6 +1440,9 @@ name = "either" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +dependencies = [ + "serde", +] [[package]] name = "elliptic-curve" @@ -1568,6 +1601,17 @@ dependencies = [ "zlib-rs", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1680,6 +1724,17 @@ dependencies = [ "num_cpus", ] +[[package]] +name = "futures-intrusive" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" +dependencies = [ + "futures-core", + "lock_api", + "parking_lot", +] + [[package]] name = "futures-io" version = "0.3.31" @@ -1947,6 +2002,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash 0.1.5", ] @@ -1965,6 +2022,15 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "hashlink" version = "0.11.0" @@ -2994,7 +3060,7 @@ dependencies = [ "futures", "futures-bounded", "futures-timer", - "hashlink", + "hashlink 0.11.0", "libp2p-core", "libp2p-identify", "libp2p-identity", @@ -3062,7 +3128,7 @@ dependencies = [ "futures", "futures-timer", "getrandom 0.2.15", - "hashlink", + "hashlink 0.11.0", "hex_fmt", "libp2p-core", "libp2p-identity", @@ -3281,7 +3347,7 @@ dependencies = [ name = "libp2p-peer-store" version = "0.1.0" dependencies = [ - "hashlink", + "hashlink 0.11.0", "libp2p", "libp2p-core", "libp2p-identity", @@ -3435,7 +3501,7 @@ dependencies = [ "bimap", "futures", "futures-timer", - "hashlink", + "hashlink 0.11.0", "libp2p-core", "libp2p-identity", "libp2p-request-response", @@ -3516,7 +3582,7 @@ dependencies = [ "futures", "futures-timer", "getrandom 0.2.15", - "hashlink", + "hashlink 0.11.0", "libp2p-core", "libp2p-identify", "libp2p-identity", @@ -3776,6 +3842,17 @@ dependencies = [ "libc", ] +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.4.15" @@ -6131,6 +6208,15 @@ dependencies = [ "sha1", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "spki" version = "0.7.3" @@ -6141,6 +6227,121 @@ dependencies = [ "der", ] +[[package]] +name = "sqlite-address-book-example" +version = "0.1.0" +dependencies = [ + "futures", + "libp2p", + "sqlx", + "tokio", + "tracing-subscriber", +] + +[[package]] +name = "sqlx" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" +dependencies = [ + "sqlx-core", + "sqlx-macros", + "sqlx-sqlite", +] + +[[package]] +name = "sqlx-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" +dependencies = [ + "base64", + "bytes", + "crc", + "crossbeam-queue", + "either", + "event-listener", + "futures-core", + "futures-intrusive", + "futures-io", + "futures-util", + "hashbrown 0.15.5", + "hashlink 0.10.0", + "indexmap 2.13.0", + "log", + "memchr", + "once_cell", + "percent-encoding", + "serde", + "sha2", + "smallvec", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tracing", + "url", +] + +[[package]] +name = "sqlx-macros" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d" +dependencies = [ + "proc-macro2", + "quote", + "sqlx-core", + "sqlx-macros-core", + "syn", +] + +[[package]] +name = "sqlx-macros-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" +dependencies = [ + "dotenvy", + "either", + "heck", + "hex", + "once_cell", + "proc-macro2", + "quote", + "serde", + "serde_json", + "sha2", + "sqlx-core", + "sqlx-sqlite", + "syn", + "tokio", + "url", +] + +[[package]] +name = "sqlx-sqlite" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea" +dependencies = [ + "atoi", + "flume", + "futures-channel", + "futures-core", + "futures-executor", + "futures-intrusive", + "futures-util", + "libsqlite3-sys", + "log", + "percent-encoding", + "serde", + "serde_urlencoded", + "sqlx-core", + "thiserror 2.0.18", + "tracing", + "url", +] + [[package]] name = "stable_deref_trait" version = "1.2.0" diff --git a/Cargo.toml b/Cargo.toml index 8846990fb65..aaea058bb9f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ members = [ "examples/ping", "examples/relay-server", "examples/rendezvous", + "examples/sqlite-address-book", "examples/stream", "examples/upnp", "hole-punching-tests", diff --git a/examples/sqlite-address-book/Cargo.toml b/examples/sqlite-address-book/Cargo.toml new file mode 100644 index 00000000000..b42b1407a3b --- /dev/null +++ b/examples/sqlite-address-book/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "sqlite-address-book-example" +version = "0.1.0" +edition.workspace = true +publish = false +license = "MIT" + +[package.metadata.release] +release = false + +[dependencies] +futures = { workspace = true } +libp2p = { path = "../../libp2p", features = ["macros", "noise", "ping", "tcp", "tokio", "yamux"] } +sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "sqlite"] } +tokio = { workspace = true, features = ["full"] } +tracing-subscriber = { workspace = true, features = ["env-filter"] } + +[lints] +workspace = true diff --git a/examples/sqlite-address-book/src/main.rs b/examples/sqlite-address-book/src/main.rs new file mode 100644 index 00000000000..73f34791396 --- /dev/null +++ b/examples/sqlite-address-book/src/main.rs @@ -0,0 +1,257 @@ +use std::{ + collections::{HashMap, VecDeque}, + convert::Infallible, + error::Error, + sync::atomic::AtomicUsize, + task::{Context, Poll, Waker}, + time::Duration, +}; + +use futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered}; +use libp2p::{ + Multiaddr, PeerId, + core::{Endpoint, transport::PortUse}, + noise, ping, + swarm::{ + ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, OutboundAddresses, SwarmEvent, + THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, dial_opts::DialOpts, dummy, + }, + tcp, yamux, +}; +use sqlx::{SqlitePool, sqlite::SqlitePoolOptions}; +use tracing_subscriber::EnvFilter; + +/// A [`NetworkBehaviour`] that resolves dial addresses from an SQLite database. +struct SqliteAddressBook { + pool: SqlitePool, + queue: VecDeque::ToSwarm, THandlerInEvent>>, + pending_addr: HashMap, + pending_tasks: FuturesUnordered)>>, + waker: Option, +} + +static TICKET: AtomicUsize = AtomicUsize::new(0); + +#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)] +struct Ticket(usize); + +impl Ticket { + fn next() -> Self { + Ticket(TICKET.fetch_add(1, std::sync::atomic::Ordering::Relaxed)) + } +} + +impl SqliteAddressBook { + fn put(&mut self, peer: PeerId, address: &Multiaddr) -> Ticket { + let ticket = Ticket::next(); + let addr = address.clone(); + let pool = self.pool.clone(); + let fut = async move { + let ret = sqlx::query("INSERT INTO peer_addresses (peer, address) VALUES (?, ?)") + .bind(peer.to_string()) + .bind(addr.to_string()) + .execute(&pool) + .await + .map(|_| ()); + (ticket, ret) + } + .boxed(); + self.pending_addr.insert(ticket, (peer, address.clone())); + self.pending_tasks.push(fut); + if let Some(waker) = self.waker.take() { + waker.wake(); + } + ticket + } + + // fn is_pending(&self, ticket: Ticket) -> bool { + // self.pending_addr.contains_key(&ticket) + // } +} + +impl NetworkBehaviour for SqliteAddressBook { + type ConnectionHandler = dummy::ConnectionHandler; + type ToSwarm = Infallible; + + fn handle_established_inbound_connection( + &mut self, + _: ConnectionId, + _: PeerId, + _: &Multiaddr, + _: &Multiaddr, + ) -> Result, ConnectionDenied> { + Ok(dummy::ConnectionHandler) + } + + fn handle_established_outbound_connection( + &mut self, + _: ConnectionId, + _: PeerId, + _: &Multiaddr, + _: Endpoint, + _: PortUse, + ) -> Result, ConnectionDenied> { + Ok(dummy::ConnectionHandler) + } + + fn handle_pending_outbound_connection( + &mut self, + _: ConnectionId, + maybe_peer: Option, + _: &[Multiaddr], + _: Endpoint, + ) -> OutboundAddresses { + let Some(peer) = maybe_peer else { + return OutboundAddresses::Ready(Ok(vec![])); + }; + + let pool = self.pool.clone(); + OutboundAddresses::Pending(Box::pin(async move { + let rows: Vec = + sqlx::query_scalar("SELECT address FROM peer_addresses WHERE peer = ?") + .bind(peer.to_string()) + .fetch_all(&pool) + .await + .map_err(ConnectionDenied::new)?; + + Ok(rows + .iter() + .filter_map(|address| address.parse().ok()) + .collect()) + })) + } + + fn on_swarm_event(&mut self, _: FromSwarm) {} + + fn on_connection_handler_event( + &mut self, + _: PeerId, + _: ConnectionId, + event: THandlerOutEvent, + ) { + match event {} + } + + fn poll( + &mut self, + cx: &mut Context<'_>, + ) -> Poll>> { + while let Poll::Ready(Some((ticket, result))) = self.pending_tasks.poll_next_unpin(cx) { + let (peer_id, address) = self.pending_addr.remove(&ticket).expect("ticket valid"); + if result.is_ok() { + self.queue + .push_back(ToSwarm::NewExternalAddrOfPeer { peer_id, address }); + } + } + + if let Some(event) = self.queue.pop_front() { + return Poll::Ready(event); + } + + self.waker = Some(cx.waker().clone()); + Poll::Pending + } +} + +#[derive(NetworkBehaviour)] +struct DialerBehaviour { + address_book: SqliteAddressBook, + ping: ping::Behaviour, +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await?; + sqlx::query("CREATE TABLE peer_addresses (peer TEXT NOT NULL, address TEXT NOT NULL)") + .execute(&pool) + .await?; + + let mut address_book = SqliteAddressBook { + pool, + queue: VecDeque::new(), + pending_addr: HashMap::new(), + pending_tasks: FuturesUnordered::new(), + waker: None, + }; + + let mut responder = libp2p::SwarmBuilder::with_new_identity() + .with_tokio() + .with_tcp( + tcp::Config::default(), + noise::Config::new, + yamux::Config::default, + )? + .with_behaviour(|_| ping::Behaviour::default())? + .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(30))) + .build(); + responder.listen_on("/ip4/127.0.0.1/tcp/0".parse()?)?; + + let responder_peer = *responder.local_peer_id(); + let responder_addr = loop { + if let SwarmEvent::NewListenAddr { address, .. } = responder.select_next_some().await { + break address; + } + }; + + println!("responder {responder_peer} listening on {responder_addr}"); + + address_book.put(responder_peer, &responder_addr); + println!("queued the responder address"); + + let mut dialer = libp2p::SwarmBuilder::with_new_identity() + .with_tokio() + .with_tcp( + tcp::Config::default(), + noise::Config::new, + yamux::Config::default, + )? + .with_behaviour(move |_| DialerBehaviour { + address_book, + ping: ping::Behaviour::default(), + })? + .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(30))) + .build(); + + dialer.dial( + DialOpts::peer_id(responder_peer) + .addresses(vec![]) + .extend_addresses_through_behaviour() + .build(), + )?; + + let timer = tokio::time::sleep(Duration::from_secs(10)); + futures::pin_mut!(timer); + loop { + tokio::select! { + _ = &mut timer => { + break; + } + _ = responder.select_next_some() => {} + event = dialer.select_next_some() => match event { + SwarmEvent::ConnectionEstablished { peer_id, .. } => { + println!("connected to {peer_id}"); + } + SwarmEvent::Behaviour(DialerBehaviourEvent::Ping(ping::Event { + peer, result, .. + })) => { + println!("ping to {peer}: {result:?}"); + } + SwarmEvent::OutgoingConnectionError { error, .. } => { + println!("dialing {responder_peer} failed: {error}"); + } + SwarmEvent::NewExternalAddrOfPeer { peer_id, address } => { + println!("{peer_id} is now reachable at {address}"); + } + _ => {} + } + } + } + Ok(()) +} diff --git a/misc/allow-block-list/src/lib.rs b/misc/allow-block-list/src/lib.rs index 32717e7071d..4482051b233 100644 --- a/misc/allow-block-list/src/lib.rs +++ b/misc/allow-block-list/src/lib.rs @@ -71,8 +71,8 @@ use std::{ use libp2p_core::{Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - CloseConnection, ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, THandler, - THandlerInEvent, THandlerOutEvent, ToSwarm, dummy, + CloseConnection, ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, + OutboundAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, dummy, }; /// A [`NetworkBehaviour`] that can act as an allow or block list. @@ -244,12 +244,10 @@ where peer: Option, _: &[Multiaddr], _: Endpoint, - ) -> Result, ConnectionDenied> { - if let Some(peer) = peer { - self.state.enforce(&peer)?; - } - - Ok(vec![]) + ) -> OutboundAddresses { + OutboundAddresses::Ready(peer.map_or(Ok(vec![]), |peer| { + self.state.enforce(&peer).map(|()| vec![]) + })) } fn handle_established_outbound_connection( diff --git a/misc/connection-limits/src/lib.rs b/misc/connection-limits/src/lib.rs index c3cb369ce7b..8563ef5fdfd 100644 --- a/misc/connection-limits/src/lib.rs +++ b/misc/connection-limits/src/lib.rs @@ -28,8 +28,8 @@ use std::{ use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - ConnectionClosed, ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, THandler, - THandlerInEvent, THandlerOutEvent, ToSwarm, + ConnectionClosed, ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, + OutboundAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{ConnectionEstablished, DialFailure, ListenFailure}, dummy, }; @@ -290,19 +290,21 @@ impl NetworkBehaviour for Behaviour { maybe_peer: Option, _: &[Multiaddr], _: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { if maybe_peer.is_some_and(|peer| self.is_bypassed(&peer)) { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); } - check_limit( + if let Err(cause) = check_limit( self.limits.max_pending_outgoing, self.pending_outbound_connections.len(), Kind::PendingOutgoing, - )?; + ) { + return OutboundAddresses::Ready(Err(cause)); + } self.pending_outbound_connections.insert(connection_id); - Ok(vec![]) + OutboundAddresses::Ready(Ok(vec![])) } fn handle_established_outbound_connection( diff --git a/misc/memory-connection-limits/src/lib.rs b/misc/memory-connection-limits/src/lib.rs index cea02c0b76c..3b0a83bff60 100644 --- a/misc/memory-connection-limits/src/lib.rs +++ b/misc/memory-connection-limits/src/lib.rs @@ -28,8 +28,8 @@ use std::{ use libp2p_core::{Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, dummy, + ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, OutboundAddresses, THandler, + THandlerInEvent, THandlerOutEvent, ToSwarm, dummy, }; use sysinfo::MemoryRefreshKind; @@ -172,9 +172,8 @@ impl NetworkBehaviour for Behaviour { _: Option, _: &[Multiaddr], _: Endpoint, - ) -> Result, ConnectionDenied> { - self.check_limit()?; - Ok(vec![]) + ) -> OutboundAddresses { + OutboundAddresses::Ready(self.check_limit().map(|()| vec![])) } fn handle_established_outbound_connection( diff --git a/misc/memory-connection-limits/tests/util/mod.rs b/misc/memory-connection-limits/tests/util/mod.rs index 949daadcb60..925145d6f01 100644 --- a/misc/memory-connection-limits/tests/util/mod.rs +++ b/misc/memory-connection-limits/tests/util/mod.rs @@ -26,8 +26,8 @@ use std::{ use libp2p_core::{Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, dummy, + ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, OutboundAddresses, THandler, + THandlerInEvent, THandlerOutEvent, ToSwarm, dummy, }; #[derive(libp2p_swarm_derive::NetworkBehaviour)] @@ -82,9 +82,9 @@ impl NetworkBehaviour _: Option, _: &[Multiaddr], _: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { self.handle_pending(); - Ok(vec![]) + OutboundAddresses::Ready(Ok(vec![])) } fn handle_established_inbound_connection( diff --git a/misc/peer-store/src/behaviour.rs b/misc/peer-store/src/behaviour.rs index 95ff93f89d2..63b16ba7db2 100644 --- a/misc/peer-store/src/behaviour.rs +++ b/misc/peer-store/src/behaviour.rs @@ -79,16 +79,16 @@ where maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: libp2p_core::Endpoint, - ) -> Result, libp2p_swarm::ConnectionDenied> { + ) -> libp2p_swarm::OutboundAddresses { if maybe_peer.is_none() { - return Ok(Vec::new()); + return libp2p_swarm::OutboundAddresses::Ready(Ok(Vec::new())); } let peer = maybe_peer.expect("already handled"); - Ok(self + libp2p_swarm::OutboundAddresses::Ready(Ok(self .store .addresses_of_peer(&peer) .map(|i| i.cloned().collect()) - .unwrap_or_default()) + .unwrap_or_default())) } fn handle_established_outbound_connection( diff --git a/protocols/autonat/src/v1/behaviour.rs b/protocols/autonat/src/v1/behaviour.rs index aefd3a337b0..cead1c392b2 100644 --- a/protocols/autonat/src/v1/behaviour.rs +++ b/protocols/autonat/src/v1/behaviour.rs @@ -39,8 +39,8 @@ use libp2p_request_response::{ self as request_response, InboundRequestId, OutboundRequestId, ProtocolSupport, ResponseChannel, }; use libp2p_swarm::{ - ConnectionDenied, ConnectionId, ListenAddresses, NetworkBehaviour, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, + ConnectionDenied, ConnectionId, ListenAddresses, NetworkBehaviour, OutboundAddresses, THandler, + THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm}, }; use web_time::Instant; @@ -505,7 +505,7 @@ impl NetworkBehaviour for Behaviour { maybe_peer: Option, addresses: &[Multiaddr], effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { self.inner.handle_pending_outbound_connection( connection_id, maybe_peer, diff --git a/protocols/identify/src/behaviour.rs b/protocols/identify/src/behaviour.rs index 36b14f40879..a55b9b286ba 100644 --- a/protocols/identify/src/behaviour.rs +++ b/protocols/identify/src/behaviour.rs @@ -34,8 +34,8 @@ use libp2p_core::{ use libp2p_identity::{Keypair, PeerId, PublicKey}; use libp2p_swarm::{ _address_translation, ConnectionDenied, ConnectionId, DialError, ExternalAddresses, - ListenAddresses, NetworkBehaviour, NotifyHandler, PeerAddresses, StreamUpgradeError, THandler, - THandlerInEvent, THandlerOutEvent, ToSwarm, + ListenAddresses, NetworkBehaviour, NotifyHandler, OutboundAddresses, PeerAddresses, + StreamUpgradeError, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm}, }; @@ -530,12 +530,12 @@ impl NetworkBehaviour for Behaviour { maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let Some(peer) = maybe_peer else { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); }; - Ok(self.discovered_peers.get(&peer)) + OutboundAddresses::Ready(Ok(self.discovered_peers.get(&peer))) } fn on_swarm_event(&mut self, event: FromSwarm) { diff --git a/protocols/kad/src/behaviour.rs b/protocols/kad/src/behaviour.rs index c1fa242d0f5..a473f52c82c 100644 --- a/protocols/kad/src/behaviour.rs +++ b/protocols/kad/src/behaviour.rs @@ -36,8 +36,8 @@ use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ ConnectionDenied, ConnectionHandler, ConnectionId, DialError, ExternalAddresses, - ListenAddresses, NetworkBehaviour, NotifyHandler, StreamProtocol, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, + ListenAddresses, NetworkBehaviour, NotifyHandler, OutboundAddresses, StreamProtocol, THandler, + THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm}, dial_opts::{self, DialOpts}, }; @@ -2252,9 +2252,9 @@ where maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let Some(peer_id) = maybe_peer else { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); }; // We should order addresses from decreasing likelihood of connectivity, so start with @@ -2276,7 +2276,7 @@ where } } - Ok(peer_addrs) + OutboundAddresses::Ready(Ok(peer_addrs)) } fn on_connection_handler_event( diff --git a/protocols/kad/src/behaviour/test.rs b/protocols/kad/src/behaviour/test.rs index 1ad415afbcb..a4cec0ad36a 100644 --- a/protocols/kad/src/behaviour/test.rs +++ b/protocols/kad/src/behaviour/test.rs @@ -31,7 +31,7 @@ use libp2p_core::{ }; use libp2p_identity as identity; use libp2p_noise as noise; -use libp2p_swarm::{self as swarm, Swarm, SwarmEvent}; +use libp2p_swarm::{self as swarm, OutboundAddresses, Swarm, SwarmEvent}; use libp2p_yamux as yamux; use quickcheck::*; use rand::{Rng, SeedableRng, random, rngs::StdRng, thread_rng}; @@ -45,6 +45,13 @@ use crate::{ type TestSwarm = Swarm>; +fn ready_addresses(outcome: OutboundAddresses) -> Vec { + match outcome { + OutboundAddresses::Ready(result) => result.expect("addresses, not a denial"), + OutboundAddresses::Pending(_) => panic!("expected synchronously-ready addresses"), + } +} + fn build_node() -> (Multiaddr, TestSwarm) { build_node_with_config(Default::default()) } @@ -1414,15 +1421,13 @@ fn network_behaviour_on_address_change() { // configured protocol name, so the peer is not yet in the // local routing table and hence no addresses are known. assert!( - kademlia - .handle_pending_outbound_connection( - connection_id, - Some(remote_peer_id), - &[], - Endpoint::Dialer - ) - .unwrap() - .is_empty() + ready_addresses(kademlia.handle_pending_outbound_connection( + connection_id, + Some(remote_peer_id), + &[], + Endpoint::Dialer + )) + .is_empty() ); // Mimic the connection handler confirming the protocol for @@ -1435,14 +1440,12 @@ fn network_behaviour_on_address_change() { assert_eq!( vec![old_address.clone()], - kademlia - .handle_pending_outbound_connection( - connection_id, - Some(remote_peer_id), - &[], - Endpoint::Dialer - ) - .unwrap(), + ready_addresses(kademlia.handle_pending_outbound_connection( + connection_id, + Some(remote_peer_id), + &[], + Endpoint::Dialer + )), ); kademlia.on_swarm_event(FromSwarm::AddressChange(AddressChange { @@ -1462,14 +1465,12 @@ fn network_behaviour_on_address_change() { assert_eq!( vec![new_address], - kademlia - .handle_pending_outbound_connection( - connection_id, - Some(remote_peer_id), - &[], - Endpoint::Dialer - ) - .unwrap(), + ready_addresses(kademlia.handle_pending_outbound_connection( + connection_id, + Some(remote_peer_id), + &[], + Endpoint::Dialer + )), ); } diff --git a/protocols/mdns/src/behaviour.rs b/protocols/mdns/src/behaviour.rs index 1dd4a379499..520d56fe8ab 100644 --- a/protocols/mdns/src/behaviour.rs +++ b/protocols/mdns/src/behaviour.rs @@ -44,8 +44,8 @@ use if_watch::IfEvent; use libp2p_core::{Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - ConnectionDenied, ConnectionId, ListenAddresses, NetworkBehaviour, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, behaviour::FromSwarm, dummy, + ConnectionDenied, ConnectionId, ListenAddresses, NetworkBehaviour, OutboundAddresses, THandler, + THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::FromSwarm, dummy, }; use smallvec::SmallVec; @@ -228,17 +228,17 @@ where maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let Some(peer_id) = maybe_peer else { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); }; - Ok(self + OutboundAddresses::Ready(Ok(self .discovered_nodes .iter() .filter(|(peer, _, _)| peer == &peer_id) .map(|(_, addr, _)| addr.clone()) - .collect()) + .collect())) } fn handle_established_outbound_connection( diff --git a/protocols/rendezvous/src/client.rs b/protocols/rendezvous/src/client.rs index 1a424bce08a..3d2f1a112d0 100644 --- a/protocols/rendezvous/src/client.rs +++ b/protocols/rendezvous/src/client.rs @@ -33,8 +33,8 @@ use libp2p_core::{Endpoint, Multiaddr, PeerRecord, transport::PortUse}; use libp2p_identity::{Keypair, PeerId, SigningError}; use libp2p_request_response::{OutboundRequestId, ProtocolSupport}; use libp2p_swarm::{ - ConnectionDenied, ConnectionId, ExternalAddresses, FromSwarm, NetworkBehaviour, THandler, - THandlerInEvent, THandlerOutEvent, ToSwarm, + ConnectionDenied, ConnectionId, ExternalAddresses, FromSwarm, NetworkBehaviour, + OutboundAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, }; use crate::codec::{ @@ -333,11 +333,11 @@ impl NetworkBehaviour for Behaviour { maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let addrs = maybe_peer .map(|peer| self.discovered_peer_addrs(&peer).cloned().collect()) .unwrap_or_default(); - Ok(addrs) + OutboundAddresses::Ready(Ok(addrs)) } } diff --git a/protocols/request-response/src/lib.rs b/protocols/request-response/src/lib.rs index 86ad6f3472e..0eb0131ea65 100644 --- a/protocols/request-response/src/lib.rs +++ b/protocols/request-response/src/lib.rs @@ -89,7 +89,7 @@ use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ ConnectionDenied, ConnectionHandler, ConnectionId, DialError, NetworkBehaviour, NotifyHandler, - PeerAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, + OutboundAddresses, PeerAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{AddressChange, ConnectionClosed, DialFailure, FromSwarm}, dial_opts::DialOpts, }; @@ -791,9 +791,9 @@ where maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let Some(peer) = maybe_peer else { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); }; let mut addresses = Vec::new(); @@ -804,7 +804,7 @@ where let cached_addrs = self.addresses.get(&peer); addresses.extend(cached_addrs); - Ok(addresses) + OutboundAddresses::Ready(Ok(addresses)) } fn handle_established_outbound_connection( diff --git a/swarm-derive/src/lib.rs b/swarm-derive/src/lib.rs index f18779d78aa..9e9a20f2677 100644 --- a/swarm-derive/src/lib.rs +++ b/swarm-derive/src/lib.rs @@ -77,6 +77,8 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> syn::Result syn::Result quote! { - combined_addresses.extend(#trait_to_impl::handle_pending_outbound_connection(&mut self.#i, connection_id, maybe_peer, addresses, effective_role)?); - }, - None => quote! { - combined_addresses.extend(#trait_to_impl::handle_pending_outbound_connection(&mut self.#field_n, connection_id, maybe_peer, addresses, effective_role)?); - } + let resolve_stmts = data_struct.fields.iter().enumerate().map(|(field_n, field)| { + let access = match field.ident { + Some(ref i) => quote! { self.#i }, + None => quote! { self.#field_n }, + }; + + quote! { + match #trait_to_impl::handle_pending_outbound_connection(&mut #access, connection_id, maybe_peer, addresses, effective_role) { + #outbound_addresses::Ready(::std::result::Result::Ok(addrs)) => { + slots.push(#either_ident::Left(addrs)); + } + #outbound_addresses::Ready(::std::result::Result::Err(denied)) => { + return #outbound_addresses::Ready(::std::result::Result::Err(denied)); + } + #outbound_addresses::Pending(fut) => { + slots.push(#either_ident::Right(fut)); } - }); + } + } + }); quote! { - let mut combined_addresses = vec![]; - - #(#extend_stmts)* - - Ok(combined_addresses) + type __Slot = #either_ident< + ::std::vec::Vec<#multiaddr>, + #box_future<'static, ::std::result::Result<::std::vec::Vec<#multiaddr>, #connection_denied>>, + >; + let mut slots: ::std::vec::Vec<__Slot> = ::std::vec::Vec::new(); + + #(#resolve_stmts)* + + if slots.iter().all(|slot| ::std::matches!(slot, #either_ident::Left(_))) { + let mut combined: ::std::vec::Vec<#multiaddr> = ::std::vec::Vec::new(); + for slot in slots { + if let #either_ident::Left(addrs) = slot { + combined.extend(addrs); + } + } + #outbound_addresses::Ready(::std::result::Result::Ok(combined)) + } else { + #outbound_addresses::Pending(::std::boxed::Box::pin(async move { + let mut combined: ::std::vec::Vec<#multiaddr> = ::std::vec::Vec::new(); + for slot in slots { + match slot { + #either_ident::Left(addrs) => combined.extend(addrs), + #either_ident::Right(fut) => combined.extend(fut.await?), + } + } + ::std::result::Result::Ok(combined) + })) + } } }; @@ -443,14 +477,13 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> syn::Result, addresses: &[#multiaddr], effective_role: #endpoint, - ) -> std::result::Result<::std::vec::Vec<#multiaddr>, #connection_denied> { + ) -> #outbound_addresses { #handle_pending_outbound_connection } diff --git a/swarm/src/behaviour.rs b/swarm/src/behaviour.rs index a333851e892..eaeab226e46 100644 --- a/swarm/src/behaviour.rs +++ b/swarm/src/behaviour.rs @@ -27,6 +27,7 @@ pub mod toggle; use std::task::{Context, Poll}; pub use external_addresses::ExternalAddresses; +use futures::future::BoxFuture; use libp2p_core::{ ConnectedPoint, Endpoint, Multiaddr, transport::{ListenerId, PortUse}, @@ -177,15 +178,22 @@ pub trait NetworkBehaviour: 'static { /// [`WithPeerIdWithAddresses::extend_addresses_through_behaviour`](crate::dial_opts::WithPeerIdWithAddresses::extend_addresses_through_behaviour) /// is set. /// - /// Any error returned from this function will immediately abort the dial attempt. + /// The addresses can be provided either synchronously ([`OutboundAddresses::Ready`], the + /// default) or asynchronously ([`OutboundAddresses::Pending`]). A + /// [`OutboundAddresses::Pending`] future is driven by the [`Swarm`](crate::Swarm) poll loop + /// and is never polled on the dialing thread, so a behaviour can source addresses from a + /// store that is not held in memory (for example an on-disk database) without blocking the + /// `Swarm`. + /// + /// Returning an `Err` (synchronously or from the future) immediately aborts the dial attempt. fn handle_pending_outbound_connection( &mut self, _connection_id: ConnectionId, _maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { - Ok(vec![]) + ) -> OutboundAddresses { + OutboundAddresses::Ready(Ok(vec![])) } /// Callback that is invoked for every established outbound connection. @@ -230,6 +238,20 @@ pub trait NetworkBehaviour: 'static { -> Poll>>; } +/// The outcome of [`NetworkBehaviour::handle_pending_outbound_connection`]. +/// +/// A behaviour returns [`OutboundAddresses::Ready`] when it can supply (or deny) the dial addresses +/// synchronously, or [`OutboundAddresses::Pending`] when it needs to resolve them asynchronously +/// without blocking the [`Swarm`](crate::Swarm). +pub enum OutboundAddresses { + /// Addresses are available now, or the dial is denied. Handled synchronously by + /// [`Swarm::dial`](crate::Swarm), preserving its synchronous `DialError` return. + Ready(Result, ConnectionDenied>), + /// Addresses must be resolved asynchronously. The future is driven to completion by the `Swarm` + /// poll loop and is never polled on the dialing thread. Returning `Err` denies the dial. + Pending(BoxFuture<'static, Result, ConnectionDenied>>), +} + /// A command issued from a [`NetworkBehaviour`] for the [`Swarm`]. /// /// [`Swarm`]: super::Swarm diff --git a/swarm/src/behaviour/either.rs b/swarm/src/behaviour/either.rs index abc79eb137e..f63839a2d6e 100644 --- a/swarm/src/behaviour/either.rs +++ b/swarm/src/behaviour/either.rs @@ -26,7 +26,7 @@ use libp2p_identity::PeerId; use crate::{ ConnectionDenied, THandler, THandlerInEvent, THandlerOutEvent, - behaviour::{self, NetworkBehaviour, ToSwarm}, + behaviour::{self, NetworkBehaviour, OutboundAddresses, ToSwarm}, connection::ConnectionId, }; @@ -82,23 +82,21 @@ where maybe_peer: Option, addresses: &[Multiaddr], effective_role: Endpoint, - ) -> Result, ConnectionDenied> { - let addresses = match self { + ) -> OutboundAddresses { + match self { Either::Left(inner) => inner.handle_pending_outbound_connection( connection_id, maybe_peer, addresses, effective_role, - )?, + ), Either::Right(inner) => inner.handle_pending_outbound_connection( connection_id, maybe_peer, addresses, effective_role, - )?, - }; - - Ok(addresses) + ), + } } fn handle_established_outbound_connection( diff --git a/swarm/src/behaviour/toggle.rs b/swarm/src/behaviour/toggle.rs index de86499b79a..def3b612ca6 100644 --- a/swarm/src/behaviour/toggle.rs +++ b/swarm/src/behaviour/toggle.rs @@ -27,7 +27,7 @@ use libp2p_identity::PeerId; use crate::{ ConnectionDenied, NetworkBehaviour, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, - behaviour::FromSwarm, + behaviour::{FromSwarm, OutboundAddresses}, connection::ConnectionId, handler::{ AddressChange, ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, @@ -118,19 +118,16 @@ where maybe_peer: Option, addresses: &[Multiaddr], effective_role: Endpoint, - ) -> Result, ConnectionDenied> { - let Some(inner) = self.inner.as_mut() else { - return Ok(vec![]); - }; - - let addresses = inner.handle_pending_outbound_connection( - connection_id, - maybe_peer, - addresses, - effective_role, - )?; - - Ok(addresses) + ) -> OutboundAddresses { + match self.inner.as_mut() { + Some(inner) => inner.handle_pending_outbound_connection( + connection_id, + maybe_peer, + addresses, + effective_role, + ), + None => OutboundAddresses::Ready(Ok(Vec::new())), + } } fn handle_established_outbound_connection( diff --git a/swarm/src/lib.rs b/swarm/src/lib.rs index 34c63944571..dfe0340859d 100644 --- a/swarm/src/lib.rs +++ b/swarm/src/lib.rs @@ -72,7 +72,7 @@ mod translation; #[doc(hidden)] pub mod derive_prelude { pub use either::Either; - pub use futures::prelude as futures; + pub use futures::{future::BoxFuture, prelude as futures}; pub use libp2p_core::{ ConnectedPoint, Endpoint, Multiaddr, transport::{ListenerId, PortUse}, @@ -86,7 +86,7 @@ pub mod derive_prelude { AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredListenAddr, ExternalAddrConfirmed, ExternalAddrExpired, FromSwarm, ListenFailure, ListenerClosed, ListenerError, NewExternalAddrCandidate, NewExternalAddrOfPeer, NewListenAddr, - NewListener, + NewListener, OutboundAddresses, }, connection::ConnectionId, }; @@ -105,7 +105,7 @@ pub use behaviour::{ AddressChange, CloseConnection, ConnectionClosed, DialFailure, ExpiredListenAddr, ExternalAddrExpired, ExternalAddresses, FromSwarm, ListenAddresses, ListenFailure, ListenerClosed, ListenerError, NetworkBehaviour, NewExternalAddrCandidate, - NewExternalAddrOfPeer, NewListenAddr, NotifyHandler, PeerAddresses, ToSwarm, + NewExternalAddrOfPeer, NewListenAddr, NotifyHandler, OutboundAddresses, PeerAddresses, ToSwarm, }; pub use connection::{ConnectionError, ConnectionId, SupportedProtocols, pool::ConnectionCounters}; use connection::{ @@ -114,16 +114,21 @@ use connection::{ }; use dial_opts::{DialOpts, PeerCondition}; pub use executor::Executor; -use futures::{prelude::*, stream::FusedStream}; +use futures::{ + future::{BoxFuture, Either}, + prelude::*, + stream::{FusedStream, FuturesUnordered}, +}; +use futures_timer::Delay; pub use handler::{ ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerSelect, OneShotHandler, OneShotHandlerConfig, StreamUpgradeError, SubstreamProtocol, }; use libp2p_core::{ - Multiaddr, Transport, + Endpoint, Multiaddr, Transport, connection::ConnectedPoint, muxing::StreamMuxerBox, - transport::{self, ListenerId, TransportError, TransportEvent}, + transport::{self, ListenerId, PortUse, TransportError, TransportEvent}, }; use libp2p_identity::PeerId; #[cfg(feature = "macros")] @@ -337,8 +342,46 @@ where pending_handler_event: Option<(PeerId, PendingNotifyHandler, THandlerInEvent)>, pending_swarm_events: VecDeque>, + + /// In-flight asynchronous outbound address resolutions. Each future is tagged with the + /// [`ConnectionId`] of its dial, polled in [`Swarm::poll_next_event`], and joined back up with + /// its [`PendingDial`] via that id. + resolving_outbound: FuturesUnordered< + BoxFuture<'static, (ConnectionId, Result, ConnectionDenied>)>, + >, + + /// Dial parameters stashed while the behaviour resolves the addresses, keyed by the + /// [`ConnectionId`] reserved for the dial. Consumed (or dropped on abort) when the matching + /// entry in [`Self::resolving_outbound`] completes. + pending_outbound_dials: HashMap, + + /// How long a behaviour's asynchronous outbound address resolution may run before the dial is + /// failed. Configured via [`Config::with_outbound_address_resolution_timeout`]. + outbound_address_resolution_timeout: Duration, +} + +/// State stashed by [`Swarm::dial`] while a [`NetworkBehaviour`] resolves the addresses of a +/// pending outgoing connection asynchronously. Used to build the transport dials once resolution +/// completes. +struct PendingDial { + /// The peer being dialed, if known. + peer_id: Option, + /// The [`PeerCondition`] re-checked when resolution completes (the peer may have connected or + /// started dialing in the meantime). + condition: PeerCondition, + /// Addresses supplied directly via [`DialOpts`]. + addresses_from_opts: Vec, + extend_addresses_through_behaviour: bool, + role_override: Endpoint, + port_use: PortUse, + dial_concurrency_override: Option, } +/// Default for [`Config::with_outbound_address_resolution_timeout`]: how long a behaviour's +/// asynchronous outbound address resolution may run before the dial is failed with +/// [`DialError::Denied`]. A safety net against a behaviour future that never resolves. +const DEFAULT_OUTBOUND_ADDRESS_RESOLUTION_TIMEOUT: Duration = Duration::from_secs(60); + impl Unpin for Swarm where TBehaviour: NetworkBehaviour {} impl Swarm @@ -365,6 +408,9 @@ where listened_addrs: HashMap::new(), pending_handler_event: None, pending_swarm_events: VecDeque::default(), + resolving_outbound: FuturesUnordered::new(), + pending_outbound_dials: HashMap::new(), + outbound_address_resolution_timeout: config.outbound_address_resolution_timeout, } } @@ -432,17 +478,7 @@ where let condition = dial_opts.peer_condition(); let connection_id = dial_opts.connection_id(); - let should_dial = match (condition, peer_id) { - (_, None) => true, - (PeerCondition::Always, _) => true, - (PeerCondition::Disconnected, Some(peer_id)) => !self.pool.is_connected(peer_id), - (PeerCondition::NotDialing, Some(peer_id)) => !self.pool.is_dialing(peer_id), - (PeerCondition::DisconnectedAndNotDialing, Some(peer_id)) => { - !self.pool.is_dialing(peer_id) && !self.pool.is_connected(peer_id) - } - }; - - if !should_dial { + if !self.should_dial(condition, peer_id) { let e = DialError::DialPeerConditionFalse(condition); self.behaviour @@ -455,64 +491,226 @@ where return Err(e); } - let addresses = { - let mut addresses_from_opts = dial_opts.get_addresses(); + let mut addresses_from_opts = dial_opts.get_addresses(); - match self.behaviour.handle_pending_outbound_connection( - connection_id, - peer_id, - addresses_from_opts.as_slice(), - dial_opts.role_override(), - ) { - Ok(addresses) => { - if dial_opts.extend_addresses_through_behaviour() { - addresses_from_opts.extend(addresses) - } else { - let num_addresses = addresses.len(); - - if num_addresses > 0 { - tracing::debug!( - connection=%connection_id, - discarded_addresses_count=%num_addresses, - "discarding addresses from `NetworkBehaviour` because `DialOpts::extend_addresses_through_behaviour is `false` for connection" - ) - } - } + // Ask the behaviour for the dial addresses. A `Ready` answer (the default, and every + // in-memory behaviour) is handled synchronously here and preserves the synchronous + // `DialError` return. A `Pending` answer defers the dial. The future is driven by + // `Swarm::poll_next_event`, never on this thread, and its eventual errors surface as + // `SwarmEvent::OutgoingConnectionError`. + match self.behaviour.handle_pending_outbound_connection( + connection_id, + peer_id, + &addresses_from_opts, + dial_opts.role_override(), + ) { + OutboundAddresses::Ready(Ok(behaviour_addresses)) => { + if dial_opts.extend_addresses_through_behaviour() { + addresses_from_opts.extend(behaviour_addresses); + } else if !behaviour_addresses.is_empty() { + tracing::debug!( + connection=%connection_id, + discarded_addresses_count=%behaviour_addresses.len(), + "discarding addresses from `NetworkBehaviour` because `DialOpts::extend_addresses_through_behaviour` is `false` for connection" + ); } - Err(cause) => { - let error = DialError::Denied { cause }; - self.behaviour - .on_swarm_event(FromSwarm::DialFailure(DialFailure { - peer_id, - error: &error, - connection_id, - })); - - return Err(error); - } + self.add_outgoing_dials( + peer_id, + connection_id, + addresses_from_opts, + dial_opts.role_override(), + dial_opts.port_use(), + dial_opts.dial_concurrency_override(), + ) } + OutboundAddresses::Ready(Err(cause)) => { + let error = DialError::Denied { cause }; - let mut unique_addresses = HashSet::new(); - addresses_from_opts.retain(|addr| { - !self.listened_addrs.values().flatten().any(|a| a == addr) - && unique_addresses.insert(addr.clone()) - }); - - if addresses_from_opts.is_empty() { - let error = DialError::NoAddresses; self.behaviour .on_swarm_event(FromSwarm::DialFailure(DialFailure { peer_id, error: &error, connection_id, })); - return Err(error); - }; - addresses_from_opts + Err(error) + } + OutboundAddresses::Pending(future) => { + // Tag the resolution with its `ConnectionId` and bound it with a timeout, then + // drive it in `Swarm::poll_next_event`. + let timeout = self.outbound_address_resolution_timeout; + self.resolving_outbound.push( + async move { + let result = + match futures::future::select(future, Delay::new(timeout)).await { + Either::Left((result, _)) => result, + Either::Right(((), _)) => Err(ConnectionDenied::new( + "outbound address resolution timed out", + )), + }; + + (connection_id, result) + } + .boxed(), + ); + + self.pending_outbound_dials.insert( + connection_id, + PendingDial { + peer_id, + condition, + addresses_from_opts, + extend_addresses_through_behaviour: dial_opts + .extend_addresses_through_behaviour(), + role_override: dial_opts.role_override(), + port_use: dial_opts.port_use(), + dial_concurrency_override: dial_opts.dial_concurrency_override(), + }, + ); + + Ok(()) + } + } + } + + /// Whether a dial to `peer_id` should proceed under `condition`, accounting for established + /// connections, in-flight dials, *and* dials whose addresses are still being resolved. + fn should_dial(&self, condition: PeerCondition, peer_id: Option) -> bool { + match (condition, peer_id) { + (_, None) => true, + (PeerCondition::Always, _) => true, + (PeerCondition::Disconnected, Some(peer_id)) => !self.pool.is_connected(peer_id), + (PeerCondition::NotDialing, Some(peer_id)) => { + !self.pool.is_dialing(peer_id) && !self.has_pending_outbound_resolution(peer_id) + } + (PeerCondition::DisconnectedAndNotDialing, Some(peer_id)) => { + !self.pool.is_dialing(peer_id) + && !self.pool.is_connected(peer_id) + && !self.has_pending_outbound_resolution(peer_id) + } + } + } + + /// Whether there is an outbound dial to `peer` whose addresses are still being resolved. + fn has_pending_outbound_resolution(&self, peer: PeerId) -> bool { + self.pending_outbound_dials + .values() + .any(|dial| dial.peer_id == Some(peer)) + } + + /// Handles completion of an asynchronous outbound address resolution started in [`Self::dial`]. + fn on_outbound_addresses_resolved( + &mut self, + connection_id: ConnectionId, + result: Result, ConnectionDenied>, + ) { + let Some(pending) = self.pending_outbound_dials.remove(&connection_id) else { + // The dial was aborted (e.g. via `disconnect`) while resolution was in flight. + return; + }; + let peer_id = pending.peer_id; + + let behaviour_addresses = match result { + Ok(addresses) => addresses, + Err(cause) => { + self.fail_deferred_dial(connection_id, peer_id, DialError::Denied { cause }); + return; + } }; + // Re-check the condition: the peer may have connected / started dialing during resolution. + if !self.should_dial(pending.condition, peer_id) { + self.fail_deferred_dial( + connection_id, + peer_id, + DialError::DialPeerConditionFalse(pending.condition), + ); + return; + } + + let mut addresses = pending.addresses_from_opts; + if pending.extend_addresses_through_behaviour { + addresses.extend(behaviour_addresses); + } else if !behaviour_addresses.is_empty() { + tracing::debug!( + connection=%connection_id, + discarded_addresses_count=%behaviour_addresses.len(), + "discarding addresses from `NetworkBehaviour` because `DialOpts::extend_addresses_through_behaviour` is `false` for connection" + ); + } + + // `add_outgoing_dials` emits `DialFailure` itself on `NoAddresses`; surface it to the + // application too, since there is no synchronous `Swarm::dial` caller to return it to. + if let Err(error) = self.add_outgoing_dials( + peer_id, + connection_id, + addresses, + pending.role_override, + pending.port_use, + pending.dial_concurrency_override, + ) { + self.pending_swarm_events + .push_back(SwarmEvent::OutgoingConnectionError { + peer_id, + connection_id, + error, + }); + } + } + + /// Reports a deferred dial failure both to the behaviour ([`FromSwarm::DialFailure`]) and the + /// application ([`SwarmEvent::OutgoingConnectionError`]). Used when the failure is detected + /// *after* the synchronous [`Swarm::dial`] call has already returned `Ok`. + fn fail_deferred_dial( + &mut self, + connection_id: ConnectionId, + peer_id: Option, + error: DialError, + ) { + self.behaviour + .on_swarm_event(FromSwarm::DialFailure(DialFailure { + peer_id, + error: &error, + connection_id, + })); + self.pending_swarm_events + .push_back(SwarmEvent::OutgoingConnectionError { + peer_id, + connection_id, + error, + }); + } + + /// Builds the transport dials for `addresses` and hands them to the [`Pool`]. Shared by the + /// synchronous and deferred [`Swarm::dial`] paths. Emits [`FromSwarm::DialFailure`] and returns + /// [`DialError::NoAddresses`] when nothing is left to dial after filtering. + fn add_outgoing_dials( + &mut self, + peer_id: Option, + connection_id: ConnectionId, + mut addresses: Vec, + role_override: Endpoint, + port_use: PortUse, + dial_concurrency_override: Option, + ) -> Result<(), DialError> { + let mut unique_addresses = HashSet::new(); + addresses.retain(|addr| { + !self.listened_addrs.values().flatten().any(|a| a == addr) + && unique_addresses.insert(addr.clone()) + }); + + if addresses.is_empty() { + let error = DialError::NoAddresses; + self.behaviour + .on_swarm_event(FromSwarm::DialFailure(DialFailure { + peer_id, + error: &error, + connection_id, + })); + return Err(error); + } + let dials = addresses .into_iter() .map(|a| match peer_id.map_or(Ok(a.clone()), |p| a.with_p2p(p)) { @@ -520,8 +718,8 @@ where let dial = self.transport.dial( address.clone(), transport::DialOpts { - role: dial_opts.role_override(), - port_use: dial_opts.port_use(), + role: role_override, + port_use, }, ); let span = tracing::debug_span!(parent: tracing::Span::none(), "Transport::dial", %address); @@ -545,9 +743,9 @@ where self.pool.add_outgoing( dials, peer_id, - dial_opts.role_override(), - dial_opts.port_use(), - dial_opts.dial_concurrency_override(), + role_override, + port_use, + dial_concurrency_override, connection_id, ); @@ -638,6 +836,12 @@ where let was_connected = self.pool.is_connected(peer_id); self.pool.disconnect(peer_id); + // Abandon any in-flight outbound dials to this peer whose addresses are still being + // resolved. The resolution futures keep running but their results are discarded once they + // land (their `pending_outbound_dials` entry is gone). + self.pending_outbound_dials + .retain(|_, dial| dial.peer_id != Some(peer_id)); + if was_connected { Ok(()) } else { Err(()) } } @@ -1235,6 +1439,16 @@ where }, } + // Drive any in-flight asynchronous outbound address resolutions. + match this.resolving_outbound.poll_next_unpin(cx) { + Poll::Ready(Some((connection_id, result))) => { + this.on_outbound_addresses_resolved(connection_id, result); + continue; + } + // `None` => the set is currently empty; `Pending` => nothing ready yet. + Poll::Ready(None) | Poll::Pending => {} + } + // Poll the known peers. match this.pool.poll(cx) { Poll::Pending => {} @@ -1374,6 +1588,7 @@ where pub struct Config { pool_config: PoolConfig, + outbound_address_resolution_timeout: Duration, } impl Config { @@ -1382,6 +1597,7 @@ impl Config { pub fn with_executor(executor: impl Executor + Send + 'static) -> Self { Self { pool_config: PoolConfig::new(Some(Box::new(executor))), + outbound_address_resolution_timeout: DEFAULT_OUTBOUND_ADDRESS_RESOLUTION_TIMEOUT, } } @@ -1390,6 +1606,7 @@ impl Config { pub fn without_executor() -> Self { Self { pool_config: PoolConfig::new(None), + outbound_address_resolution_timeout: DEFAULT_OUTBOUND_ADDRESS_RESOLUTION_TIMEOUT, } } @@ -1505,6 +1722,20 @@ impl Config { self.pool_config.idle_connection_timeout = timeout; self } + + /// Configures how long a [`NetworkBehaviour`]'s asynchronous outbound address resolution + /// (see [`NetworkBehaviour::handle_pending_outbound_connection`]) may run before the dial is + /// failed with [`DialError::Denied`]. + /// + /// This only applies to behaviours that return [`OutboundAddresses::Pending`]. Resolutions that + /// complete synchronously ([`OutboundAddresses::Ready`], the default) are never timed out. It + /// is a safety net against a behaviour future that never resolves. + /// + /// Defaults to 60 seconds. + pub fn with_outbound_address_resolution_timeout(mut self, timeout: Duration) -> Self { + self.outbound_address_resolution_timeout = timeout; + self + } } /// Possible errors when trying to establish or upgrade an outbound connection. diff --git a/swarm/src/test.rs b/swarm/src/test.rs index e03bf591227..490590e7859 100644 --- a/swarm/src/test.rs +++ b/swarm/src/test.rs @@ -31,8 +31,8 @@ use libp2p_core::{ use libp2p_identity::PeerId; use crate::{ - ConnectionDenied, ConnectionHandler, ConnectionId, NetworkBehaviour, THandler, THandlerInEvent, - THandlerOutEvent, ToSwarm, + ConnectionDenied, ConnectionHandler, ConnectionId, NetworkBehaviour, OutboundAddresses, + THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::{ ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredListenAddr, ExternalAddrExpired, FromSwarm, ListenerClosed, ListenerError, NewExternalAddrCandidate, @@ -112,12 +112,12 @@ where maybe_peer: Option, _addresses: &[Multiaddr], _effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { let Some(p) = maybe_peer else { - return Ok(vec![]); + return OutboundAddresses::Ready(Ok(vec![])); }; - Ok(self.addresses.get(&p).map_or(Vec::new(), |v| v.clone())) + OutboundAddresses::Ready(Ok(self.addresses.get(&p).map_or(Vec::new(), |v| v.clone()))) } fn poll(&mut self, _: &mut Context<'_>) -> Poll>> { @@ -416,7 +416,7 @@ where maybe_peer: Option, addresses: &[Multiaddr], effective_role: Endpoint, - ) -> Result, ConnectionDenied> { + ) -> OutboundAddresses { self.handle_pending_outbound_connection.push(( maybe_peer, addresses.to_vec(), diff --git a/swarm/tests/deferred_dial.rs b/swarm/tests/deferred_dial.rs new file mode 100644 index 00000000000..fd392fc0b04 --- /dev/null +++ b/swarm/tests/deferred_dial.rs @@ -0,0 +1,239 @@ +use std::{ + task::{Context, Poll}, + time::Duration, +}; + +use futures::{FutureExt, channel::oneshot}; +use libp2p_core::{ + Endpoint, Multiaddr, Transport, + transport::{PortUse, dummy::DummyTransport}, +}; +use libp2p_identity::{Keypair, PeerId}; +use libp2p_swarm::{ + Config, ConnectionDenied, ConnectionId, DialError, FromSwarm, NetworkBehaviour, + OutboundAddresses, Swarm, SwarmEvent, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, + dial_opts::DialOpts, dummy, +}; +use libp2p_swarm_test::SwarmExt; + +/// A [`NetworkBehaviour`] that resolves the addresses for an outbound dial **asynchronously** +/// ([`OutboundAddresses::Pending`]), so the dial always takes the deferred, `Swarm::poll`-driven +/// path rather than the synchronous fast-path. +/// +/// If `gate` is set, the resolution future blocks until the gate is released, which lets a test +/// abort the dial mid-resolution. +struct AsyncAddrBehaviour { + addr: Multiaddr, + gate: Option>, +} + +impl NetworkBehaviour for AsyncAddrBehaviour { + type ConnectionHandler = dummy::ConnectionHandler; + type ToSwarm = std::convert::Infallible; + + fn handle_established_inbound_connection( + &mut self, + _: ConnectionId, + _: PeerId, + _: &Multiaddr, + _: &Multiaddr, + ) -> Result, ConnectionDenied> { + Ok(dummy::ConnectionHandler) + } + + fn handle_established_outbound_connection( + &mut self, + _: ConnectionId, + _: PeerId, + _: &Multiaddr, + _: Endpoint, + _: PortUse, + ) -> Result, ConnectionDenied> { + Ok(dummy::ConnectionHandler) + } + + fn handle_pending_outbound_connection( + &mut self, + _: ConnectionId, + _: Option, + _: &[Multiaddr], + _: Endpoint, + ) -> OutboundAddresses { + let addr = self.addr.clone(); + let gate = self.gate.take(); + + OutboundAddresses::Pending( + async move { + match gate { + // Block until the test releases the gate. + Some(gate) => { + let _ = gate.await; + } + // Yield once so the deferred path is still exercised (never `Ready` inline). + None => tokio::task::yield_now().await, + } + Ok(vec![addr]) + } + .boxed(), + ) + } + + fn on_swarm_event(&mut self, _: FromSwarm) {} + + fn on_connection_handler_event( + &mut self, + _: PeerId, + _: ConnectionId, + event: THandlerOutEvent, + ) { + // `dummy::ConnectionHandler` never emits an event (`ToBehaviour = Infallible`). + match event {} + } + + fn poll(&mut self, _: &mut Context<'_>) -> Poll>> { + Poll::Pending + } +} + +/// Dialing by `PeerId` only, with the address supplied asynchronously by the behaviour, still +/// establishes the connection, exercising the deferred (`Pending`) address-resolution path. +#[tokio::test] +async fn deferred_outbound_address_resolution_dials_resolved_address() { + let mut responder = Swarm::new_ephemeral_tokio(|_| dummy::Behaviour); + let (responder_addr, _) = responder.listen().with_memory_addr_external().await; + let responder_peer = *responder.local_peer_id(); + + let mut dialer = Swarm::new_ephemeral_tokio(|_| AsyncAddrBehaviour { + addr: responder_addr, + gate: None, + }); + + dialer + .dial( + DialOpts::peer_id(responder_peer) + .addresses(vec![]) + .extend_addresses_through_behaviour() + .build(), + ) + .expect("`Swarm::dial` to accept the deferred dial"); + + loop { + tokio::select! { + _ = responder.next_swarm_event() => {} + event = dialer.next_swarm_event() => match event { + SwarmEvent::ConnectionEstablished { peer_id, .. } => { + assert_eq!(peer_id, responder_peer); + break; + } + SwarmEvent::OutgoingConnectionError { error, .. } => { + panic!("deferred dial failed: {error:?}"); + } + _ => {} + } + } + } +} + +/// Aborting a peer (`disconnect_peer_id`) while its dial addresses are still being resolved +/// abandons the dial: no connection is established and no error is surfaced, even after the +/// resolution future eventually completes. +#[tokio::test] +async fn aborting_during_resolution_abandons_the_dial() { + let mut responder = Swarm::new_ephemeral_tokio(|_| dummy::Behaviour); + let (responder_addr, _) = responder.listen().with_memory_addr_external().await; + let responder_peer = *responder.local_peer_id(); + + let (release_gate, gate) = oneshot::channel(); + let mut dialer = Swarm::new_ephemeral_tokio(|_| AsyncAddrBehaviour { + addr: responder_addr, + gate: Some(gate), + }); + + dialer + .dial( + DialOpts::peer_id(responder_peer) + .addresses(vec![]) + .extend_addresses_through_behaviour() + .build(), + ) + .expect("`Swarm::dial` to accept the deferred dial"); + + // Abort the dial while resolution is still gated, then let resolution complete. + let _ = dialer.disconnect_peer_id(responder_peer); + release_gate.send(()).unwrap(); + + // Drive both swarms: the resolved addresses must be discarded, so no terminal event for the + // responder should ever arrive. We expect the timeout to elapse. + let outcome = tokio::time::timeout(Duration::from_millis(300), async { + loop { + tokio::select! { + _ = responder.next_swarm_event() => {} + event = dialer.next_swarm_event() => match event { + SwarmEvent::ConnectionEstablished { peer_id, .. } if peer_id == responder_peer => { + return "connected"; + } + SwarmEvent::OutgoingConnectionError { peer_id: Some(p), .. } if p == responder_peer => { + return "errored"; + } + _ => {} + } + } + } + }) + .await; + + assert!( + outcome.is_err(), + "aborted dial should produce neither a connection nor an error, got {outcome:?}" + ); +} + +/// A resolution that never completes is failed once the configured +/// [`Config::with_outbound_address_resolution_timeout`] elapses. +#[tokio::test] +async fn outbound_address_resolution_times_out() { + let keypair = Keypair::generate_ed25519(); + let local_peer_id = keypair.public().to_peer_id(); + + // Hold the sender for the whole test so the gate is never released: resolution never completes. + let (_release_gate, gate) = oneshot::channel(); + + let mut dialer = Swarm::new( + DummyTransport::new().boxed(), + AsyncAddrBehaviour { + addr: "/memory/1".parse().unwrap(), + gate: Some(gate), + }, + local_peer_id, + Config::with_tokio_executor() + .with_outbound_address_resolution_timeout(Duration::from_millis(100)), + ); + + let target = PeerId::random(); + dialer + .dial( + DialOpts::peer_id(target) + .addresses(vec![]) + .extend_addresses_through_behaviour() + .build(), + ) + .expect("`Swarm::dial` to accept the deferred dial"); + + let (peer_id, error) = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let SwarmEvent::OutgoingConnectionError { peer_id, error, .. } = + dialer.next_swarm_event().await + { + return (peer_id, error); + } + } + }) + .await + .expect("the resolution timeout should fail the dial"); + + assert_eq!(peer_id, Some(target)); + assert!( + matches!(error, DialError::Denied { .. }), + "expected `DialError::Denied` from the resolution timeout, got {error:?}" + ); +}