Merge branch 'master' into kpp-quic

client/cli/src/params/network_params.rs

@@ -72,7 +72,7 @@ pub struct NetworkParams {
 	#[clap(long)]
 	pub experimental_quic: bool,
 
-	/// Always forbid connecting to private IPv4 addresses (as specified in
+	/// Always forbid connecting to private IPv4/IPv6 addresses (as specified in
 	/// [RFC1918](https://tools.ietf.org/html/rfc1918)), unless the address was passed with
 	/// `--reserved-nodes` or `--bootnodes`. Enabled by default for chains marked as "live" in
 	/// their chain specifications.

client/network/Cargo.toml

@@ -25,9 +25,7 @@ fnv = "1.0.6"
 futures = "0.3.21"
 futures-timer = "3.0.2"
 ip_network = "0.4.1"
-libp2p = { version = "0.50.0", features = ["async-std", "dns", "identify", "kad", "macros", "mdns", "mplex", "noise", "ping", "quic", "tcp", "yamux", "websocket"] }
-linked_hash_set = "0.1.3"
-linked-hash-map = "0.5.4"
+libp2p = { version = "0.50.0", features = ["dns", "identify", "kad", "macros", "mdns", "mplex", "noise", "ping", "quic", "tcp",  "tokio", "yamux", "websocket"] }
 log = "0.4.17"
 lru = "0.8.1"
 parking_lot = "0.12.1"

client/network/src/discovery.rs

@@ -61,7 +61,7 @@ use libp2p::{
 		GetClosestPeersError, GetRecordOk, Kademlia, KademliaBucketInserts, KademliaConfig,
 		KademliaEvent, QueryId, QueryResult, Quorum, Record,
 	},
-	mdns::{async_io::Behaviour as Mdns, Config as MdnsConfig, Event as MdnsEvent},
+	mdns::{self, tokio::Behaviour as TokioMdns},
 	multiaddr::Protocol,
 	swarm::{
 		behaviour::{
@@ -249,7 +249,7 @@ impl DiscoveryConfig {
 				NonZeroUsize::new(MAX_KNOWN_EXTERNAL_ADDRESSES)
 					.expect("value is a constant; constant is non-zero; qed."),
 			),
-			outbound_query_records: Vec::new(),
+			records_to_publish: Default::default(),
 		}
 	}
 }
@@ -287,8 +287,12 @@ pub struct DiscoveryBehaviour {
 	allow_non_globals_in_dht: bool,
 	/// A cache of discovered external addresses. Only used for logging purposes.
 	known_external_addresses: LruHashSet<Multiaddr>,
-	/// A cache of outbound query records.
-	outbound_query_records: Vec<(record::Key, Vec<u8>)>,
+	/// Records to publish per QueryId.
+	///
+	/// After finishing a Kademlia query, libp2p will return us a list of the closest peers that
+	/// did not return the record(in `FinishedWithNoAdditionalRecord`). We will then put the record
+	/// to these peers.
+	records_to_publish: HashMap<QueryId, Record>,
 }
 
 impl DiscoveryBehaviour {
@@ -544,8 +548,8 @@ impl NetworkBehaviour for DiscoveryBehaviour {
 			FromSwarm::ListenerClosed(e) => {
 				self.kademlia.on_swarm_event(FromSwarm::ListenerClosed(e));
 			},
-			FromSwarm::ListenFailure(_) => {
-				// NetworkBehaviour::inject_listen_failure on Kademlia<MemoryStore> does nothing.
+			FromSwarm::ListenFailure(e) => {
+				self.kademlia.on_swarm_event(FromSwarm::ListenFailure(e));
 			},
 			FromSwarm::ListenerError(e) => {
 				self.kademlia.on_swarm_event(FromSwarm::ListenerError(e));
@@ -692,33 +696,57 @@ impl NetworkBehaviour for DiscoveryBehaviour {
 					KademliaEvent::OutboundQueryProgressed {
 						result: QueryResult::GetRecord(res),
 						stats,
-						step,
+						id,
 						..
 					} => {
 						let ev = match res {
-							Ok(ok) =>
-								if let GetRecordOk::FoundRecord(r) = ok {
-									self.outbound_query_records
-										.push((r.record.key, r.record.value));
-									continue
-								} else {
-									debug!(
-										target: "sub-libp2p",
-										"Libp2p => Query progressed to {:?} step (last: {:?})",
-										step.count,
-										step.last,
-									);
-									if step.last {
-										let records =
-											self.outbound_query_records.drain(..).collect();
-										DiscoveryOut::ValueFound(
-											records,
-											stats.duration().unwrap_or_default(),
-										)
-									} else {
+							Ok(GetRecordOk::FoundRecord(r)) => {
+								debug!(
+									target: "sub-libp2p",
+									"Libp2p => Found record ({:?}) with value: {:?}",
+									r.record.key,
+									r.record.value,
+								);
+
+								// Let's directly finish the query, as we are only interested in a
+								// quorum of 1.
+								if let Some(kad) = self.kademlia.as_mut() {
+									if let Some(mut query) = kad.query_mut(&id) {
+										query.finish();
+									}
+								}
+
+								// Will be removed below when we receive
+								// `FinishedWithNoAdditionalRecord`.
+								self.records_to_publish.insert(id, r.record.clone());
+
+								DiscoveryOut::ValueFound(
+									vec![(r.record.key, r.record.value)],
+									stats.duration().unwrap_or_default(),
+								)
+							},
+							Ok(GetRecordOk::FinishedWithNoAdditionalRecord {
+								cache_candidates,
+							}) => {
+								// We always need to remove the record to not leak any data!
+								if let Some(record) = self.records_to_publish.remove(&id) {
+									if cache_candidates.is_empty() {
 										continue
 									}
-								},
+
+									// Put the record to the `cache_candidates` that are nearest to
+									// the record key from our point of view of the network.
+									if let Some(kad) = self.kademlia.as_mut() {
+										kad.put_record_to(
+											record,
+											cache_candidates.into_iter().map(|v| v.1),
+											Quorum::One,
+										);
+									}
+								}
+
+								continue
+							},
 							Err(e @ libp2p::kad::GetRecordError::NotFound { .. }) => {
 								trace!(
 									target: "sub-libp2p",
@@ -927,10 +955,12 @@ mod tests {
 					config.finish()
 				};
 
-				let mut swarm = Swarm::with_threadpool_executor(
+				let runtime = tokio::runtime::Runtime::new().unwrap();
+				let mut swarm = Swarm::with_executor(
 					transport,
 					behaviour,
 					keypair.public().to_peer_id(),
+					TokioExecutor(runtime),
 				);
 				let listen_addr: Multiaddr =
 					format!("/memory/{}", rand::random::<u64>()).parse().unwrap();

client/network/src/protocol/notifications/tests.rs

@@ -25,13 +25,14 @@ use libp2p::{
 	core::{connection::ConnectionId, transport::MemoryTransport, upgrade},
 	identity, noise,
 	swarm::{
-		behaviour::FromSwarm, ConnectionHandler, IntoConnectionHandler, NetworkBehaviour,
+		behaviour::FromSwarm, ConnectionHandler, Executor, IntoConnectionHandler, NetworkBehaviour,
 		NetworkBehaviourAction, PollParameters, Swarm, SwarmEvent,
 	},
 	yamux, Multiaddr, PeerId, Transport,
 };
 use std::{
 	iter,
+	pin::Pin,
 	task::{Context, Poll},
 	time::Duration,
 };
@@ -103,10 +104,12 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 				.collect(),
 		};
 
-		let mut swarm = Swarm::with_threadpool_executor(
+		let runtime = tokio::runtime::Runtime::new().unwrap();
+		let mut swarm = Swarm::with_executor(
 			transport,
 			behaviour,
 			keypairs[index].public().to_peer_id(),
+			TokioExecutor(runtime),
 		);
 		swarm.listen_on(addrs[index].clone()).unwrap();
 		out.push(swarm);

client/network/src/request_responses.rs

@@ -344,13 +344,21 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 						)
 					}
 				},
-			FromSwarm::DialFailure(DialFailure { peer_id, error, .. }) =>
-				for (p, _) in self.protocols.values_mut() {
-					let handler = p.new_handler();
-					NetworkBehaviour::on_swarm_event(
-						p,
-						FromSwarm::DialFailure(DialFailure { peer_id, handler, error }),
-					);
+			FromSwarm::DialFailure(DialFailure { peer_id, error, handler }) =>
+				for (p_name, p_handler) in handler.into_iter() {
+					if let Some((proto, _)) = self.protocols.get_mut(p_name.as_str()) {
+						proto.on_swarm_event(FromSwarm::DialFailure(DialFailure {
+							peer_id,
+							handler: p_handler,
+							error,
+						}));
+					} else {
+						log::error!(
+						  target: "sub-libp2p",
+						  "on_swarm_event/dial_failure: no request-response instance registered for protocol {:?}",
+						  p_name,
+						)
+					}
 				},
 			FromSwarm::ListenerClosed(e) =>
 				for (p, _) in self.protocols.values_mut() {
@@ -414,9 +422,11 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 			return proto.on_connection_handler_event(peer_id, connection_id, event)
 		}
 
-		log::warn!(target: "sub-libp2p",
-      "on_connection_handler_event: no request-response instance registered for protocol {:?}",
-      p_name)
+		log::warn!(
+			target: "sub-libp2p",
+			"on_connection_handler_event: no request-response instance registered for protocol {:?}",
+			p_name
+		);
 	}
 
 	fn poll(
@@ -951,8 +961,13 @@ mod tests {
 
 		let behaviour = RequestResponsesBehaviour::new(list, handle).unwrap();
 
-		let mut swarm =
-			Swarm::with_threadpool_executor(transport, behaviour, keypair.public().to_peer_id());
+		let runtime = tokio::runtime::Runtime::new().unwrap();
+		let mut swarm = Swarm::with_executor(
+			transport,
+			behaviour,
+			keypair.public().to_peer_id(),
+			TokioExecutor(runtime),
+		);
 		let listen_addr: Multiaddr = format!("/memory/{}", rand::random::<u64>()).parse().unwrap();
 
 		swarm.listen_on(listen_addr.clone()).unwrap();

client/network/src/service.rs

@@ -38,7 +38,7 @@ use crate::{
 	transport, ChainSyncInterface, ReputationChange,
 };
 
-use futures::{channel::oneshot, executor::ThreadPoolBuilder, prelude::*};
+use futures::{channel::oneshot, prelude::*};
 use libp2p::{
 	core::{either::EitherError, upgrade, ConnectedPoint},
 	identify::Info as IdentifyInfo,
@@ -375,17 +375,19 @@ where
 				}
 			};
 
-			let builder = if let Some(spawner) = params.executor {
+			let builder = {
 				struct SpawnImpl<F>(F);
 				impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for SpawnImpl<F> {
 					fn exec(&self, f: Pin<Box<dyn Future<Output = ()> + Send>>) {
 						(self.0)(f)
 					}
 				}
-				SwarmBuilder::with_executor(transport, behaviour, local_peer_id, SpawnImpl(spawner))
-			} else {
-				let tp = ThreadPoolBuilder::new().name_prefix("libp2p-swarm-task-").create()?;
-				SwarmBuilder::with_executor(transport, behaviour, local_peer_id, tp)
+				SwarmBuilder::with_executor(
+					transport,
+					behaviour,
+					local_peer_id,
+					SpawnImpl(params.executor),
+				)
 			};
 			let builder = builder
 				.connection_limits(
@@ -399,6 +401,7 @@ where
 				.notify_handler_buffer_size(NonZeroUsize::new(32).expect("32 != 0; qed"))
 				.connection_event_buffer_size(1024)
 				.max_negotiating_inbound_streams(2048);
+
 			(builder.build(), bandwidth)
 		};
 

client/network/src/transport.rs

@@ -40,7 +40,7 @@ fn build_quic_transport(
 	keypair: &identity::Keypair,
 ) -> Boxed<(PeerId, StreamMuxerBox)> {
 	let config = QuicConfig::new(&keypair);
-	let transport = QuicTransport::<quic::async_std::Provider>::new(config)
+	let transport = QuicTransport::<quic::tokio::Provider>::new(config)
 		.map(|(p, c), _| (p, StreamMuxerBox::new(c)))
 		.boxed();
 
@@ -71,11 +71,11 @@ pub fn build_transport(
 ) -> (Boxed<(PeerId, StreamMuxerBox)>, Arc<BandwidthSinks>) {
 	// Build the base layer of the transport.
 	let transport = if !memory_only {
+		// Main transport: DNS(TCP)
 		let tcp_config = tcp::Config::new().nodelay(true);
-		let desktop_trans = tcp::async_io::Transport::new(tcp_config.clone());
-		let desktop_trans = websocket::WsConfig::new(desktop_trans)
-			.or_transport(tcp::async_io::Transport::new(tcp_config.clone()));
-		let dns_init = futures::executor::block_on(dns::DnsConfig::system(desktop_trans));
+		let tcp_trans = tcp::tokio::Transport::new(tcp_config.clone());
+		let dns_init = dns::TokioDnsConfig::system(tcp_trans);
+
 		EitherTransport::Left(if let Ok(dns) = dns_init {
 			// WS + WSS transport
 			//
@@ -87,10 +87,11 @@ pub fn build_transport(
 				.expect("same system_conf & resolver to work");
 			EitherTransport::Left(websocket::WsConfig::new(dns_for_wss).or_transport(dns))
 		} else {
-			let desktop_trans = tcp::async_io::Transport::new(tcp_config.clone());
-			let desktop_trans = websocket::WsConfig::new(desktop_trans)
-				.or_transport(tcp::async_io::Transport::new(tcp_config));
-			EitherTransport::Right(desktop_trans.map_err(dns::DnsErr::Transport))
+			// In case DNS can't be constructed, fallback to TCP + WS (WSS won't work)
+			let tcp_trans = tcp::tokio::Transport::new(tcp_config.clone());
+			let desktop_trans = websocket::WsConfig::new(tcp_trans)
+				.or_transport(tcp::tokio::Transport::new(tcp_config));
+			EitherTransport::Right(desktop_trans)
 		})
 	} else {
 		EitherTransport::Right(OptionalTransport::some(

client/network/transactions/Cargo.toml

@@ -16,7 +16,6 @@ targets = ["x86_64-unknown-linux-gnu"]
 array-bytes = "4.1"
 codec = { package = "parity-scale-codec", version = "3.2.2", features = ["derive"] }
 futures = "0.3.21"
-hex = "0.4.0"
 libp2p = "0.50.0"
 log = "0.4.17"
 pin-project = "1.0.12"

client/telemetry/Cargo.toml

@@ -16,7 +16,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 [dependencies]
 chrono = "0.4.19"
 futures = "0.3.21"
-libp2p = { version = "0.50.0", features = ["async-std", "dns", "tcp", "wasm-ext", "websocket"] }
+libp2p = { version = "0.50.0", features = ["dns", "tcp", "tokio", "wasm-ext", "websocket"] }
 log = "0.4.17"
 parking_lot = "0.12.1"
 pin-project = "1.0.12"

client/telemetry/src/transport.rs

@@ -30,8 +30,8 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
 
 pub(crate) fn initialize_transport() -> Result<WsTrans, io::Error> {
 	let transport = {
-		let tcp_transport = libp2p::tcp::async_io::Transport::new(libp2p::tcp::Config::new());
-		let inner = block_on(libp2p::dns::DnsConfig::system(tcp_transport))?;
+		let tcp_transport = libp2p::tcp::tokio::Transport::new(libp2p::tcp::Config::new());
+		let inner = libp2p::dns::TokioDnsConfig::system(tcp_transport)?;
 		libp2p::websocket::framed::WsConfig::new(inner).and_then(|connec, _| {
 			let connec = connec
 				.with(|item| {