Upgrade libp2p to 0.49.0 (#12256)

* cargo upgrade libp2p * Get rid of `NetworkBehaviourEventProcess` in handling of `CustomMessageOutcome` * Get rid of `NetworkBehaviourEventProcess` in handling of `request_responses::Event` * Get rid of `NetworkBehaviourEventProcess` in handling of `peer_info::PeerInfoEvent` * Get rid of `NetworkBehaviourEventProcess` in handling of `DiscoveryOut` * Get rid of `poll()` method in `Bahaviour` * minor: comments * Upgrade libp2p to 0.49.0 (unreleased) * Support multiple Kad protocol names * Make borrow checker happy * minor: wording * Make substrate build with libp2p-0.49.0 * rustfmt * Get rid of MdnsWrapper * Resolve deprecation warnings * Fix documentation * Apply suggestions from code review: fix typos Co-authored-by: Aaro Altonen <48052676+altonen@users.noreply.github.com> * Apply suggestion: simplify kad protocol name matching Co-authored-by: Aaro Altonen <48052676+altonen@users.noreply.github.com>
2026-05-10 01:47:58 +00:00 · 2022-10-17 14:02:37 +03:00
parent 882023f570
commit bb175f0373
21 changed files with 543 additions and 846 deletions
@@ -24,18 +24,13 @@ use crate::{
 };

 use bytes::Bytes;
-use codec::Encode;
 use futures::channel::oneshot;
 use libp2p::{
 	core::{Multiaddr, PeerId, PublicKey},
-	identify::IdentifyInfo,
+	identify::Info as IdentifyInfo,
 	kad::record,
-	swarm::{
-		NetworkBehaviour, NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters,
-	},
 	NetworkBehaviour,
 };
-use log::debug;

 use sc_consensus::import_queue::{IncomingBlock, RuntimeOrigin};
 use sc_network_common::{
@@ -46,26 +41,22 @@ use sc_network_common::{
 		ProtocolName,
 	},
 	request_responses::{IfDisconnected, ProtocolConfig, RequestFailure},
+	sync::{warp::WarpProofRequest, OpaqueBlockRequest, OpaqueStateRequest},
 };
-use sc_peerset::PeersetHandle;
+use sc_peerset::{PeersetHandle, ReputationChange};
 use sp_blockchain::HeaderBackend;
 use sp_consensus::BlockOrigin;
 use sp_runtime::{
 	traits::{Block as BlockT, NumberFor},
 	Justifications,
 };
-use std::{
-	collections::{HashSet, VecDeque},
-	iter,
-	task::{Context, Poll},
-	time::Duration,
-};
+use std::{collections::HashSet, time::Duration};

 pub use crate::request_responses::{InboundFailure, OutboundFailure, RequestId, ResponseFailure};

 /// General behaviour of the network. Combines all protocols together.
 #[derive(NetworkBehaviour)]
-#[behaviour(out_event = "BehaviourOut<B>", poll_method = "poll", event_process = true)]
+#[behaviour(out_event = "BehaviourOut<B>")]
 pub struct Behaviour<B, Client>
 where
 	B: BlockT,
@@ -80,25 +71,6 @@ where
 	discovery: DiscoveryBehaviour,
 	/// Generic request-response protocols.
 	request_responses: request_responses::RequestResponsesBehaviour,
-
-	/// Queue of events to produce for the outside.
-	#[behaviour(ignore)]
-	events: VecDeque<BehaviourOut<B>>,
-
-	/// Protocol name used to send out block requests via
-	/// [`request_responses::RequestResponsesBehaviour`].
-	#[behaviour(ignore)]
-	block_request_protocol_name: String,
-
-	/// Protocol name used to send out state requests via
-	/// [`request_responses::RequestResponsesBehaviour`].
-	#[behaviour(ignore)]
-	state_request_protocol_name: String,
-
-	/// Protocol name used to send out warp sync requests via
-	/// [`request_responses::RequestResponsesBehaviour`].
-	#[behaviour(ignore)]
-	warp_sync_protocol_name: Option<String>,
 }

 /// Event generated by `Behaviour`.
@@ -107,7 +79,7 @@ pub enum BehaviourOut<B: BlockT> {
 	JustificationImport(RuntimeOrigin, B::Hash, NumberFor<B>, Justifications),

 	/// Started a random iterative Kademlia discovery query.
-	RandomKademliaStarted(ProtocolId),
+	RandomKademliaStarted(Vec<ProtocolId>),

 	/// We have received a request from a peer and answered it.
 	///
@@ -136,6 +108,12 @@ pub enum BehaviourOut<B: BlockT> {
 		result: Result<(), RequestFailure>,
 	},

+	/// A request protocol handler issued reputation changes for the given peer.
+	ReputationChanges {
+		peer: PeerId,
+		changes: Vec<ReputationChange>,
+	},
+
 	/// Opened a substream with the given node with the given notifications protocol.
 	///
 	/// The protocol is always one of the notification protocols that have been registered.
@@ -186,15 +164,60 @@ pub enum BehaviourOut<B: BlockT> {
 		messages: Vec<(ProtocolName, Bytes)>,
 	},

+	/// A new block request must be emitted.
+	BlockRequest {
+		/// Node we send the request to.
+		target: PeerId,
+		/// Opaque implementation-specific block request.
+		request: OpaqueBlockRequest,
+		/// One-shot channel to receive the response.
+		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+	},
+
+	/// A new state request must be emitted.
+	StateRequest {
+		/// Node we send the request to.
+		target: PeerId,
+		/// Opaque implementation-specific state request.
+		request: OpaqueStateRequest,
+		/// One-shot channel to receive the response.
+		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+	},
+
+	/// A new warp sync request must be emitted.
+	WarpSyncRequest {
+		/// Node we send the request to.
+		target: PeerId,
+		/// Warp sync request.
+		request: WarpProofRequest<B>,
+		/// One-shot channel to receive the response.
+		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+	},
+
 	/// Now connected to a new peer for syncing purposes.
 	SyncConnected(PeerId),

 	/// No longer connected to a peer for syncing purposes.
 	SyncDisconnected(PeerId),

+	/// We have obtained identity information from a peer, including the addresses it is listening
+	/// on.
+	PeerIdentify {
+		/// Id of the peer that has been identified.
+		peer_id: PeerId,
+		/// Information about the peer.
+		info: IdentifyInfo,
+	},
+
+	/// We have learned about the existence of a node on the default set.
+	Discovered(PeerId),
+
 	/// Events generated by a DHT as a response to get_value or put_value requests as well as the
 	/// request duration.
 	Dht(DhtEvent, Duration),
+
+	/// Ignored event generated by lower layers.
+	None,
 }

 impl<B, Client> Behaviour<B, Client>
@@ -216,17 +239,9 @@ where
 		mut request_response_protocols: Vec<ProtocolConfig>,
 		peerset: PeersetHandle,
 	) -> Result<Self, request_responses::RegisterError> {
-		// Extract protocol name and add to `request_response_protocols`.
-		let block_request_protocol_name = block_request_protocol_config.name.to_string();
-		let state_request_protocol_name = state_request_protocol_config.name.to_string();
-		let warp_sync_protocol_name = match warp_sync_protocol_config {
-			Some(config) => {
-				let name = config.name.to_string();
-				request_response_protocols.push(config);
-				Some(name)
-			},
-			None => None,
-		};
+		if let Some(config) = warp_sync_protocol_config {
+			request_response_protocols.push(config);
+		}
 		request_response_protocols.push(block_request_protocol_config);
 		request_response_protocols.push(state_request_protocol_config);
 		request_response_protocols.push(light_client_request_protocol_config);
@@ -239,10 +254,6 @@ where
 				request_response_protocols.into_iter(),
 				peerset,
 			)?,
-			events: VecDeque::new(),
-			block_request_protocol_name,
-			state_request_protocol_name,
-			warp_sync_protocol_name,
 		})
 	}

@@ -310,6 +321,17 @@ where
 		&mut self.substrate
 	}

+	/// Add a self-reported address of a remote peer to the k-buckets of the supported
+	/// DHTs (`supported_protocols`).
+	pub fn add_self_reported_address_to_dht(
+		&mut self,
+		peer_id: &PeerId,
+		supported_protocols: &[impl AsRef<[u8]>],
+		addr: Multiaddr,
+	) {
+		self.discovery.add_self_reported_address(peer_id, supported_protocols, addr);
+	}
+
 	/// Start querying a record from the DHT. Will later produce either a `ValueFound` or a
 	/// `ValueNotFound` event.
 	pub fn get_value(&mut self, key: record::Key) {
@@ -333,221 +355,91 @@ fn reported_roles_to_observed_role(roles: Roles) -> ObservedRole {
 	}
 }

-impl<B, Client> NetworkBehaviourEventProcess<CustomMessageOutcome<B>> for Behaviour<B, Client>
-where
-	B: BlockT,
-	Client: HeaderBackend<B> + 'static,
-{
-	fn inject_event(&mut self, event: CustomMessageOutcome<B>) {
+impl<B: BlockT> From<CustomMessageOutcome<B>> for BehaviourOut<B> {
+	fn from(event: CustomMessageOutcome<B>) -> Self {
 		match event {
 			CustomMessageOutcome::BlockImport(origin, blocks) =>
-				self.events.push_back(BehaviourOut::BlockImport(origin, blocks)),
-			CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) => self
-				.events
-				.push_back(BehaviourOut::JustificationImport(origin, hash, nb, justification)),
-			CustomMessageOutcome::BlockRequest { target, request, pending_response } => {
-				match self.substrate.encode_block_request(&request) {
-					Ok(data) => {
-						self.request_responses.send_request(
-							&target,
-							&self.block_request_protocol_name,
-							data,
-							pending_response,
-							IfDisconnected::ImmediateError,
-						);
-					},
-					Err(err) => {
-						log::warn!(
-							target: "sync",
-							"Failed to encode block request {:?}: {:?}",
-							request, err
-						);
-					},
-				}
-			},
-			CustomMessageOutcome::StateRequest { target, request, pending_response } => {
-				match self.substrate.encode_state_request(&request) {
-					Ok(data) => {
-						self.request_responses.send_request(
-							&target,
-							&self.state_request_protocol_name,
-							data,
-							pending_response,
-							IfDisconnected::ImmediateError,
-						);
-					},
-					Err(err) => {
-						log::warn!(
-							target: "sync",
-							"Failed to encode state request {:?}: {:?}",
-							request, err
-						);
-					},
-				}
-			},
+				BehaviourOut::BlockImport(origin, blocks),
+			CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) =>
+				BehaviourOut::JustificationImport(origin, hash, nb, justification),
+			CustomMessageOutcome::BlockRequest { target, request, pending_response } =>
+				BehaviourOut::BlockRequest { target, request, pending_response },
+			CustomMessageOutcome::StateRequest { target, request, pending_response } =>
+				BehaviourOut::StateRequest { target, request, pending_response },
 			CustomMessageOutcome::WarpSyncRequest { target, request, pending_response } =>
-				match &self.warp_sync_protocol_name {
-					Some(name) => self.request_responses.send_request(
-						&target,
-						name,
-						request.encode(),
-						pending_response,
-						IfDisconnected::ImmediateError,
-					),
-					None => {
-						log::warn!(
-							target: "sync",
-							"Trying to send warp sync request when no protocol is configured {:?}",
-							request,
-						);
-					},
-				},
+				BehaviourOut::WarpSyncRequest { target, request, pending_response },
 			CustomMessageOutcome::NotificationStreamOpened {
 				remote,
 				protocol,
 				negotiated_fallback,
 				roles,
 				notifications_sink,
-			} => {
-				self.events.push_back(BehaviourOut::NotificationStreamOpened {
-					remote,
-					protocol,
-					negotiated_fallback,
-					role: reported_roles_to_observed_role(roles),
-					notifications_sink,
-				});
+			} => BehaviourOut::NotificationStreamOpened {
+				remote,
+				protocol,
+				negotiated_fallback,
+				role: reported_roles_to_observed_role(roles),
+				notifications_sink,
 			},
 			CustomMessageOutcome::NotificationStreamReplaced {
 				remote,
 				protocol,
 				notifications_sink,
-			} => self.events.push_back(BehaviourOut::NotificationStreamReplaced {
-				remote,
-				protocol,
-				notifications_sink,
-			}),
-			CustomMessageOutcome::NotificationStreamClosed { remote, protocol } => self
-				.events
-				.push_back(BehaviourOut::NotificationStreamClosed { remote, protocol }),
-			CustomMessageOutcome::NotificationsReceived { remote, messages } => {
-				self.events.push_back(BehaviourOut::NotificationsReceived { remote, messages });
-			},
-			CustomMessageOutcome::PeerNewBest(_peer_id, _number) => {},
-			CustomMessageOutcome::SyncConnected(peer_id) =>
-				self.events.push_back(BehaviourOut::SyncConnected(peer_id)),
+			} => BehaviourOut::NotificationStreamReplaced { remote, protocol, notifications_sink },
+			CustomMessageOutcome::NotificationStreamClosed { remote, protocol } =>
+				BehaviourOut::NotificationStreamClosed { remote, protocol },
+			CustomMessageOutcome::NotificationsReceived { remote, messages } =>
+				BehaviourOut::NotificationsReceived { remote, messages },
+			CustomMessageOutcome::PeerNewBest(_peer_id, _number) => BehaviourOut::None,
+			CustomMessageOutcome::SyncConnected(peer_id) => BehaviourOut::SyncConnected(peer_id),
 			CustomMessageOutcome::SyncDisconnected(peer_id) =>
-				self.events.push_back(BehaviourOut::SyncDisconnected(peer_id)),
-			CustomMessageOutcome::None => {},
+				BehaviourOut::SyncDisconnected(peer_id),
+			CustomMessageOutcome::None => BehaviourOut::None,
 		}
 	}
 }

-impl<B, Client> NetworkBehaviourEventProcess<request_responses::Event> for Behaviour<B, Client>
-where
-	B: BlockT,
-	Client: HeaderBackend<B> + 'static,
-{
-	fn inject_event(&mut self, event: request_responses::Event) {
+impl<B: BlockT> From<request_responses::Event> for BehaviourOut<B> {
+	fn from(event: request_responses::Event) -> Self {
 		match event {
-			request_responses::Event::InboundRequest { peer, protocol, result } => {
-				self.events.push_back(BehaviourOut::InboundRequest { peer, protocol, result });
-			},
-			request_responses::Event::RequestFinished { peer, protocol, duration, result } => {
-				self.events.push_back(BehaviourOut::RequestFinished {
-					peer,
-					protocol,
-					duration,
-					result,
-				});
-			},
+			request_responses::Event::InboundRequest { peer, protocol, result } =>
+				BehaviourOut::InboundRequest { peer, protocol, result },
+			request_responses::Event::RequestFinished { peer, protocol, duration, result } =>
+				BehaviourOut::RequestFinished { peer, protocol, duration, result },
 			request_responses::Event::ReputationChanges { peer, changes } =>
-				for change in changes {
-					self.substrate.report_peer(peer, change);
-				},
+				BehaviourOut::ReputationChanges { peer, changes },
 		}
 	}
 }

-impl<B, Client> NetworkBehaviourEventProcess<peer_info::PeerInfoEvent> for Behaviour<B, Client>
-where
-	B: BlockT,
-	Client: HeaderBackend<B> + 'static,
-{
-	fn inject_event(&mut self, event: peer_info::PeerInfoEvent) {
-		let peer_info::PeerInfoEvent::Identified {
-			peer_id,
-			info: IdentifyInfo { protocol_version, agent_version, mut listen_addrs, protocols, .. },
-		} = event;
-
-		if listen_addrs.len() > 30 {
-			debug!(
-				target: "sub-libp2p",
-				"Node {:?} has reported more than 30 addresses; it is identified by {:?} and {:?}",
-				peer_id, protocol_version, agent_version
-			);
-			listen_addrs.truncate(30);
-		}
-
-		for addr in listen_addrs {
-			self.discovery.add_self_reported_address(&peer_id, protocols.iter(), addr);
-		}
-		self.substrate.add_default_set_discovered_nodes(iter::once(peer_id));
+impl<B: BlockT> From<peer_info::PeerInfoEvent> for BehaviourOut<B> {
+	fn from(event: peer_info::PeerInfoEvent) -> Self {
+		let peer_info::PeerInfoEvent::Identified { peer_id, info } = event;
+		BehaviourOut::PeerIdentify { peer_id, info }
 	}
 }

-impl<B, Client> NetworkBehaviourEventProcess<DiscoveryOut> for Behaviour<B, Client>
-where
-	B: BlockT,
-	Client: HeaderBackend<B> + 'static,
-{
-	fn inject_event(&mut self, out: DiscoveryOut) {
-		match out {
+impl<B: BlockT> From<DiscoveryOut> for BehaviourOut<B> {
+	fn from(event: DiscoveryOut) -> Self {
+		match event {
 			DiscoveryOut::UnroutablePeer(_peer_id) => {
 				// Obtaining and reporting listen addresses for unroutable peers back
 				// to Kademlia is handled by the `Identify` protocol, part of the
-				// `PeerInfoBehaviour`. See the `NetworkBehaviourEventProcess`
-				// implementation for `PeerInfoEvent`.
-			},
-			DiscoveryOut::Discovered(peer_id) => {
-				self.substrate.add_default_set_discovered_nodes(iter::once(peer_id));
-			},
-			DiscoveryOut::ValueFound(results, duration) => {
-				self.events
-					.push_back(BehaviourOut::Dht(DhtEvent::ValueFound(results), duration));
-			},
-			DiscoveryOut::ValueNotFound(key, duration) => {
-				self.events.push_back(BehaviourOut::Dht(DhtEvent::ValueNotFound(key), duration));
-			},
-			DiscoveryOut::ValuePut(key, duration) => {
-				self.events.push_back(BehaviourOut::Dht(DhtEvent::ValuePut(key), duration));
-			},
-			DiscoveryOut::ValuePutFailed(key, duration) => {
-				self.events
-					.push_back(BehaviourOut::Dht(DhtEvent::ValuePutFailed(key), duration));
+				// `PeerInfoBehaviour`. See the `From<peer_info::PeerInfoEvent>`
+				// implementation.
+				BehaviourOut::None
 			},
+			DiscoveryOut::Discovered(peer_id) => BehaviourOut::Discovered(peer_id),
+			DiscoveryOut::ValueFound(results, duration) =>
+				BehaviourOut::Dht(DhtEvent::ValueFound(results), duration),
+			DiscoveryOut::ValueNotFound(key, duration) =>
+				BehaviourOut::Dht(DhtEvent::ValueNotFound(key), duration),
+			DiscoveryOut::ValuePut(key, duration) =>
+				BehaviourOut::Dht(DhtEvent::ValuePut(key), duration),
+			DiscoveryOut::ValuePutFailed(key, duration) =>
+				BehaviourOut::Dht(DhtEvent::ValuePutFailed(key), duration),
 			DiscoveryOut::RandomKademliaStarted(protocols) =>
-				for protocol in protocols {
-					self.events.push_back(BehaviourOut::RandomKademliaStarted(protocol));
-				},
+				BehaviourOut::RandomKademliaStarted(protocols),
 		}
 	}
 }
-
-impl<B, Client> Behaviour<B, Client>
-where
-	B: BlockT,
-	Client: HeaderBackend<B> + 'static,
-{
-	fn poll(
-		&mut self,
-		_cx: &mut Context,
-		_: &mut impl PollParameters,
-	) -> Poll<NetworkBehaviourAction<BehaviourOut<B>, <Self as NetworkBehaviour>::ConnectionHandler>>
-	{
-		if let Some(event) = self.events.pop_front() {
-			return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event))
-		}
-
-		Poll::Pending
-	}
-}
@@ -455,11 +455,8 @@ mod tests {
 	}

 	fn secret_bytes(kp: &Keypair) -> Vec<u8> {
-		match kp {
-			Keypair::Ed25519(p) => p.secret().as_ref().iter().cloned().collect(),
-			Keypair::Secp256k1(p) => p.secret().to_bytes().to_vec(),
-			_ => panic!("Unexpected keypair."),
-		}
+		let Keypair::Ed25519(p) = kp;
+		p.secret().as_ref().iter().cloned().collect()
 	}

 	#[test]
@@ -198,7 +198,7 @@ impl DiscoveryConfig {
 				let proto_name = protocol_name_from_protocol_id(&protocol_id);

 				let mut config = KademliaConfig::default();
-				config.set_protocol_name(proto_name);
+				config.set_protocol_names(std::iter::once(proto_name.into()).collect());
 				// By default Kademlia attempts to insert all peers into its routing table once a
 				// dialing attempt succeeds. In order to control which peer is added, disable the
 				// auto-insertion and instead add peers manually.
@@ -232,9 +232,15 @@ impl DiscoveryConfig {
 			allow_private_ipv4,
 			discovery_only_if_under_num,
 			mdns: if enable_mdns {
-				MdnsWrapper::Instantiating(Mdns::new(MdnsConfig::default()).boxed())
+				match Mdns::new(MdnsConfig::default()) {
+					Ok(mdns) => Some(mdns),
+					Err(err) => {
+						warn!(target: "sub-libp2p", "Failed to initialize mDNS: {:?}", err);
+						None
+					},
+				}
 			} else {
-				MdnsWrapper::Disabled
+				None
 			},
 			allow_non_globals_in_dht,
 			known_external_addresses: LruHashSet::new(
@@ -256,7 +262,7 @@ pub struct DiscoveryBehaviour {
 	/// Kademlia requests and answers.
 	kademlias: HashMap<ProtocolId, Kademlia<MemoryStore>>,
 	/// Discovers nodes on the local network.
-	mdns: MdnsWrapper,
+	mdns: Option<Mdns>,
 	/// Stream that fires when we need to perform the next random Kademlia query. `None` if
 	/// random walking is disabled.
 	next_kad_random_query: Option<Delay>,
@@ -320,7 +326,7 @@ impl DiscoveryBehaviour {
 	pub fn add_self_reported_address(
 		&mut self,
 		peer_id: &PeerId,
-		supported_protocols: impl Iterator<Item = impl AsRef<[u8]>>,
+		supported_protocols: &[impl AsRef<[u8]>],
 		addr: Multiaddr,
 	) {
 		if !self.allow_non_globals_in_dht && !self.can_add_to_dht(&addr) {
@@ -329,17 +335,18 @@ impl DiscoveryBehaviour {
 		}

 		let mut added = false;
-		for protocol in supported_protocols {
-			for kademlia in self.kademlias.values_mut() {
-				if protocol.as_ref() == kademlia.protocol_name() {
-					trace!(
-						target: "sub-libp2p",
-						"Adding self-reported address {} from {} to Kademlia DHT {}.",
-						addr, peer_id, String::from_utf8_lossy(kademlia.protocol_name()),
-					);
-					kademlia.add_address(peer_id, addr.clone());
-					added = true;
-				}
+		for kademlia in self.kademlias.values_mut() {
+			if let Some(matching_protocol) = supported_protocols
+				.iter()
+				.find(|p| kademlia.protocol_names().iter().any(|k| k.as_ref() == p.as_ref()))
+			{
+				trace!(
+					target: "sub-libp2p",
+					"Adding self-reported address {} from {} to Kademlia DHT {}.",
+					addr, peer_id, String::from_utf8_lossy(matching_protocol.as_ref()),
+				);
+				kademlia.add_address(peer_id, addr.clone());
+				added = true;
 			}
 		}

@@ -532,7 +539,9 @@ impl NetworkBehaviour for DiscoveryBehaviour {
 				list_to_filter.extend(k.addresses_of_peer(peer_id))
 			}

-			list_to_filter.extend(self.mdns.addresses_of_peer(peer_id));
+			if let Some(ref mut mdns) = self.mdns {
+				list_to_filter.extend(mdns.addresses_of_peer(peer_id));
+			}

 			if !self.allow_private_ipv4 {
 				list_to_filter.retain(|addr| match addr.iter().next() {
@@ -913,36 +922,38 @@ impl NetworkBehaviour for DiscoveryBehaviour {
 		}

 		// Poll mDNS.
-		while let Poll::Ready(ev) = self.mdns.poll(cx, params) {
-			match ev {
-				NetworkBehaviourAction::GenerateEvent(event) => match event {
-					MdnsEvent::Discovered(list) => {
-						if self.num_connections >= self.discovery_only_if_under_num {
-							continue
-						}
+		if let Some(ref mut mdns) = self.mdns {
+			while let Poll::Ready(ev) = mdns.poll(cx, params) {
+				match ev {
+					NetworkBehaviourAction::GenerateEvent(event) => match event {
+						MdnsEvent::Discovered(list) => {
+							if self.num_connections >= self.discovery_only_if_under_num {
+								continue
+							}

-						self.pending_events
-							.extend(list.map(|(peer_id, _)| DiscoveryOut::Discovered(peer_id)));
-						if let Some(ev) = self.pending_events.pop_front() {
-							return Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev))
-						}
+							self.pending_events
+								.extend(list.map(|(peer_id, _)| DiscoveryOut::Discovered(peer_id)));
+							if let Some(ev) = self.pending_events.pop_front() {
+								return Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev))
+							}
+						},
+						MdnsEvent::Expired(_) => {},
 					},
-					MdnsEvent::Expired(_) => {},
-				},
-				NetworkBehaviourAction::Dial { .. } => {
-					unreachable!("mDNS never dials!");
-				},
-				NetworkBehaviourAction::NotifyHandler { event, .. } => match event {}, /* `event` is an enum with no variant */
-				NetworkBehaviourAction::ReportObservedAddr { address, score } =>
-					return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr {
-						address,
-						score,
-					}),
-				NetworkBehaviourAction::CloseConnection { peer_id, connection } =>
-					return Poll::Ready(NetworkBehaviourAction::CloseConnection {
-						peer_id,
-						connection,
-					}),
+					NetworkBehaviourAction::Dial { .. } => {
+						unreachable!("mDNS never dials!");
+					},
+					NetworkBehaviourAction::NotifyHandler { event, .. } => match event {}, /* `event` is an enum with no variant */
+					NetworkBehaviourAction::ReportObservedAddr { address, score } =>
+						return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr {
+							address,
+							score,
+						}),
+					NetworkBehaviourAction::CloseConnection { peer_id, connection } =>
+						return Poll::Ready(NetworkBehaviourAction::CloseConnection {
+							peer_id,
+							connection,
+						}),
+				}
 			}
 		}

@@ -959,45 +970,6 @@ fn protocol_name_from_protocol_id(id: &ProtocolId) -> Vec<u8> {
 	v
 }

-/// [`Mdns::new`] returns a future. Instead of forcing [`DiscoveryConfig::finish`] and all its
-/// callers to be async, lazily instantiate [`Mdns`].
-enum MdnsWrapper {
-	Instantiating(futures::future::BoxFuture<'static, std::io::Result<Mdns>>),
-	Ready(Mdns),
-	Disabled,
-}
-
-impl MdnsWrapper {
-	fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
-		match self {
-			Self::Instantiating(_) => Vec::new(),
-			Self::Ready(mdns) => mdns.addresses_of_peer(peer_id),
-			Self::Disabled => Vec::new(),
-		}
-	}
-
-	fn poll(
-		&mut self,
-		cx: &mut Context<'_>,
-		params: &mut impl PollParameters,
-	) -> Poll<NetworkBehaviourAction<MdnsEvent, <Mdns as NetworkBehaviour>::ConnectionHandler>> {
-		loop {
-			match self {
-				Self::Instantiating(fut) =>
-					*self = match futures::ready!(fut.as_mut().poll(cx)) {
-						Ok(mdns) => Self::Ready(mdns),
-						Err(err) => {
-							warn!(target: "sub-libp2p", "Failed to initialize mDNS: {:?}", err);
-							Self::Disabled
-						},
-					},
-				Self::Ready(mdns) => return mdns.poll(cx, params),
-				Self::Disabled => return Poll::Pending,
-			}
-		}
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::{protocol_name_from_protocol_id, DiscoveryConfig, DiscoveryOut};
@@ -1100,8 +1072,7 @@ mod tests {
 												.behaviour_mut()
 												.add_self_reported_address(
 													&other,
-													[protocol_name_from_protocol_id(&protocol_id)]
-														.iter(),
+													&[protocol_name_from_protocol_id(&protocol_id)],
 													addr,
 												);

@@ -1156,7 +1127,7 @@ mod tests {
 		// Add remote peer with unsupported protocol.
 		discovery.add_self_reported_address(
 			&remote_peer_id,
-			[protocol_name_from_protocol_id(&unsupported_protocol_id)].iter(),
+			&[protocol_name_from_protocol_id(&unsupported_protocol_id)],
 			remote_addr.clone(),
 		);

@@ -1173,7 +1144,7 @@ mod tests {
 		// Add remote peer with supported protocol.
 		discovery.add_self_reported_address(
 			&remote_peer_id,
-			[protocol_name_from_protocol_id(&supported_protocol_id)].iter(),
+			&[protocol_name_from_protocol_id(&supported_protocol_id)],
 			remote_addr.clone(),
 		);

@@ -1212,7 +1183,7 @@ mod tests {
 		// Add remote peer with `protocol_a` only.
 		discovery.add_self_reported_address(
 			&remote_peer_id,
-			[protocol_name_from_protocol_id(&protocol_a)].iter(),
+			&[protocol_name_from_protocol_id(&protocol_a)],
 			remote_addr.clone(),
 		);

@@ -23,8 +23,11 @@ use libp2p::{
 		connection::ConnectionId, either::EitherOutput, transport::ListenerId, ConnectedPoint,
 		PeerId, PublicKey,
 	},
-	identify::{Identify, IdentifyConfig, IdentifyEvent, IdentifyInfo},
-	ping::{Ping, PingConfig, PingEvent, PingSuccess},
+	identify::{
+		Behaviour as Identify, Config as IdentifyConfig, Event as IdentifyEvent,
+		Info as IdentifyInfo,
+	},
+	ping::{Behaviour as Ping, Config as PingConfig, Event as PingEvent, Success as PingSuccess},
 	swarm::{
 		ConnectionHandler, IntoConnectionHandler, IntoConnectionHandlerSelect, NetworkBehaviour,
 		NetworkBehaviourAction, PollParameters,
@@ -38,9 +38,11 @@ use crate::{
 	transport, ChainSyncInterface, ReputationChange,
 };

+use codec::Encode;
 use futures::{channel::oneshot, prelude::*};
 use libp2p::{
 	core::{either::EitherError, upgrade, ConnectedPoint, Executor},
+	identify::Info as IdentifyInfo,
 	kad::record::Key as KademliaKey,
 	multiaddr,
 	ping::Failure as PingFailure,
@@ -264,6 +266,11 @@ where
 		let num_connected = Arc::new(AtomicUsize::new(0));
 		let is_major_syncing = Arc::new(AtomicBool::new(false));

+		let block_request_protocol_name = params.block_request_protocol_config.name.clone();
+		let state_request_protocol_name = params.state_request_protocol_config.name.clone();
+		let warp_sync_protocol_name =
+			params.warp_sync_protocol_config.as_ref().map(|c| c.name.clone());
+
 		// Build the swarm.
 		let (mut swarm, bandwidth): (Swarm<Behaviour<B, Client>>, _) = {
 			let user_agent = format!(
@@ -455,6 +462,9 @@ where
 			peers_notifications_sinks,
 			metrics,
 			boot_node_ids,
+			block_request_protocol_name,
+			state_request_protocol_name,
+			warp_sync_protocol_name,
 			_marker: Default::default(),
 		})
 	}
@@ -1273,6 +1283,15 @@ where
 	/// For each peer and protocol combination, an object that allows sending notifications to
 	/// that peer. Shared with the [`NetworkService`].
 	peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ProtocolName), NotificationsSink>>>,
+	/// Protocol name used to send out block requests via
+	/// [`crate::request_responses::RequestResponsesBehaviour`].
+	block_request_protocol_name: ProtocolName,
+	/// Protocol name used to send out state requests via
+	/// [`crate::request_responses::RequestResponsesBehaviour`].
+	state_request_protocol_name: ProtocolName,
+	/// Protocol name used to send out warp sync requests via
+	/// [`crate::request_responses::RequestResponsesBehaviour`].
+	warp_sync_protocol_name: Option<ProtocolName>,
 	/// Marker to pin the `H` generic. Serves no purpose except to not break backwards
 	/// compatibility.
 	_marker: PhantomData<H>,
@@ -1451,6 +1470,84 @@ where
 					}
 					this.import_queue.import_justifications(origin, hash, nb, justifications);
 				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::BlockRequest {
+					target,
+					request,
+					pending_response,
+				})) => {
+					match this
+						.network_service
+						.behaviour()
+						.user_protocol()
+						.encode_block_request(&request)
+					{
+						Ok(data) => {
+							this.network_service.behaviour_mut().send_request(
+								&target,
+								&this.block_request_protocol_name,
+								data,
+								pending_response,
+								IfDisconnected::ImmediateError,
+							);
+						},
+						Err(err) => {
+							log::warn!(
+								target: "sync",
+								"Failed to encode block request {:?}: {:?}",
+								request, err
+							);
+						},
+					}
+				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::StateRequest {
+					target,
+					request,
+					pending_response,
+				})) => {
+					match this
+						.network_service
+						.behaviour()
+						.user_protocol()
+						.encode_state_request(&request)
+					{
+						Ok(data) => {
+							this.network_service.behaviour_mut().send_request(
+								&target,
+								&this.state_request_protocol_name,
+								data,
+								pending_response,
+								IfDisconnected::ImmediateError,
+							);
+						},
+						Err(err) => {
+							log::warn!(
+								target: "sync",
+								"Failed to encode state request {:?}: {:?}",
+								request, err
+							);
+						},
+					}
+				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::WarpSyncRequest {
+					target,
+					request,
+					pending_response,
+				})) => match &this.warp_sync_protocol_name {
+					Some(name) => this.network_service.behaviour_mut().send_request(
+						&target,
+						&name,
+						request.encode(),
+						pending_response,
+						IfDisconnected::ImmediateError,
+					),
+					None => {
+						log::warn!(
+							target: "sync",
+							"Trying to send warp sync request when no protocol is configured {:?}",
+							request,
+						);
+					},
+				},
 				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::InboundRequest {
 					protocol,
 					result,
@@ -1526,14 +1623,58 @@ where
 							},
 						}
 					},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::ReputationChanges {
+					peer,
+					changes,
+				})) =>
+					for change in changes {
+						this.network_service.behaviour().user_protocol().report_peer(peer, change);
+					},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::PeerIdentify {
+					peer_id,
+					info:
+						IdentifyInfo {
+							protocol_version,
+							agent_version,
+							mut listen_addrs,
+							protocols,
+							..
+						},
+				})) => {
+					if listen_addrs.len() > 30 {
+						debug!(
+							target: "sub-libp2p",
+							"Node {:?} has reported more than 30 addresses; it is identified by {:?} and {:?}",
+							peer_id, protocol_version, agent_version
+						);
+						listen_addrs.truncate(30);
+					}
+					for addr in listen_addrs {
+						this.network_service
+							.behaviour_mut()
+							.add_self_reported_address_to_dht(&peer_id, &protocols, addr);
+					}
+					this.network_service
+						.behaviour_mut()
+						.user_protocol_mut()
+						.add_default_set_discovered_nodes(iter::once(peer_id));
+				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::Discovered(peer_id))) => {
+					this.network_service
+						.behaviour_mut()
+						.user_protocol_mut()
+						.add_default_set_discovered_nodes(iter::once(peer_id));
+				},
 				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::RandomKademliaStarted(
-					protocol,
+					protocols,
 				))) =>
 					if let Some(metrics) = this.metrics.as_ref() {
-						metrics
-							.kademlia_random_queries_total
-							.with_label_values(&[protocol.as_ref()])
-							.inc();
+						for protocol in protocols {
+							metrics
+								.kademlia_random_queries_total
+								.with_label_values(&[protocol.as_ref()])
+								.inc();
+						}
 					},
 				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamOpened {
 					remote,
@@ -1654,6 +1795,9 @@ where

 					this.event_streams.send(Event::Dht(event));
 				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::None)) => {
+					// Ignored event from lower layers.
+				},
 				Poll::Ready(SwarmEvent::ConnectionEstablished {
 					peer_id,
 					endpoint,