Revert "chore: update libp2p to 0.52.1 (#14429)" (#14722)

* Revert "chore: update libp2p to 0.52.1 (#14429)"

This reverts commit 59d8b86450.

* Fix dependencies

* Update dependencies

* Update Cargo.lock

substrate/client/network/src/protocol/notifications/behaviour.rs

@@ -986,7 +986,7 @@ impl Notifications {
 
 impl NetworkBehaviour for Notifications {
 	type ConnectionHandler = NotifsHandler;
-	type ToSwarm = NotificationsOut;
+	type OutEvent = NotificationsOut;
 
 	fn handle_pending_inbound_connection(
 		&mut self,
@@ -1459,11 +1459,10 @@ impl NetworkBehaviour for Notifications {
 			FromSwarm::ListenerClosed(_) => {},
 			FromSwarm::ListenFailure(_) => {},
 			FromSwarm::ListenerError(_) => {},
-			FromSwarm::ExternalAddrExpired(_) => {},
+			FromSwarm::ExpiredExternalAddr(_) => {},
 			FromSwarm::NewListener(_) => {},
 			FromSwarm::ExpiredListenAddr(_) => {},
-			FromSwarm::NewExternalAddrCandidate(_) => {},
-			FromSwarm::ExternalAddrConfirmed(_) => {},
+			FromSwarm::NewExternalAddr(_) => {},
 			FromSwarm::AddressChange(_) => {},
 			FromSwarm::NewListenAddr(_) => {},
 		}
@@ -2002,7 +2001,7 @@ impl NetworkBehaviour for Notifications {
 		&mut self,
 		cx: &mut Context,
 		_params: &mut impl PollParameters,
-	) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
+	) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
 		if let Some(event) = self.events.pop_front() {
 			return Poll::Ready(event)
 		}
@@ -2108,6 +2107,7 @@ mod tests {
 		protocol::notifications::handler::tests::*,
 		protocol_controller::{IncomingIndex, ProtoSetConfig, ProtocolController},
 	};
+	use libp2p::swarm::AddressRecord;
 	use sc_utils::mpsc::tracing_unbounded;
 	use std::{collections::HashSet, iter};
 
@@ -2127,14 +2127,31 @@ mod tests {
 	}
 
 	#[derive(Clone)]
-	struct MockPollParams {}
+	struct MockPollParams {
+		peer_id: PeerId,
+		addr: Multiaddr,
+	}
 
 	impl PollParameters for MockPollParams {
 		type SupportedProtocolsIter = std::vec::IntoIter<Vec<u8>>;
+		type ListenedAddressesIter = std::vec::IntoIter<Multiaddr>;
+		type ExternalAddressesIter = std::vec::IntoIter<AddressRecord>;
 
 		fn supported_protocols(&self) -> Self::SupportedProtocolsIter {
 			vec![].into_iter()
 		}
+
+		fn listened_addresses(&self) -> Self::ListenedAddressesIter {
+			vec![self.addr.clone()].into_iter()
+		}
+
+		fn external_addresses(&self) -> Self::ExternalAddressesIter {
+			vec![].into_iter()
+		}
+
+		fn local_peer_id(&self) -> &PeerId {
+			&self.peer_id
+		}
 	}
 
 	fn development_notifs() -> (Notifications, ProtocolController) {
@@ -2999,7 +3016,7 @@ mod tests {
 
 		notif.on_swarm_event(FromSwarm::DialFailure(libp2p::swarm::behaviour::DialFailure {
 			peer_id: Some(peer),
-			error: &libp2p::swarm::DialError::Aborted,
+			error: &libp2p::swarm::DialError::Banned,
 			connection_id: ConnectionId::new_unchecked(1337),
 		}));
 
@@ -3536,7 +3553,7 @@ mod tests {
 		let now = Instant::now();
 		notif.on_swarm_event(FromSwarm::DialFailure(libp2p::swarm::behaviour::DialFailure {
 			peer_id: Some(peer),
-			error: &libp2p::swarm::DialError::Aborted,
+			error: &libp2p::swarm::DialError::Banned,
 			connection_id: ConnectionId::new_unchecked(0),
 		}));
 
@@ -3656,7 +3673,7 @@ mod tests {
 		assert!(notif.peers.get(&(peer, set_id)).is_some());
 
 		if tokio::time::timeout(Duration::from_secs(5), async {
-			let mut params = MockPollParams {};
+			let mut params = MockPollParams { peer_id: PeerId::random(), addr: Multiaddr::empty() };
 
 			loop {
 				futures::future::poll_fn(|cx| {
@@ -3765,7 +3782,7 @@ mod tests {
 		// verify that the code continues to keep the peer disabled by resetting the timer
 		// after the first one expired.
 		if tokio::time::timeout(Duration::from_secs(5), async {
-			let mut params = MockPollParams {};
+			let mut params = MockPollParams { peer_id: PeerId::random(), addr: Multiaddr::empty() };
 
 			loop {
 				futures::future::poll_fn(|cx| {

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

@@ -160,7 +160,7 @@ impl std::ops::DerefMut for CustomProtoWithAddr {
 
 impl NetworkBehaviour for CustomProtoWithAddr {
 	type ConnectionHandler = <Notifications as NetworkBehaviour>::ConnectionHandler;
-	type ToSwarm = <Notifications as NetworkBehaviour>::ToSwarm;
+	type OutEvent = <Notifications as NetworkBehaviour>::OutEvent;
 
 	fn handle_pending_inbound_connection(
 		&mut self,
@@ -238,10 +238,9 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		&mut self,
 		cx: &mut Context,
 		params: &mut impl PollParameters,
-	) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
+	) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
 		let _ = self.peer_store_future.poll_unpin(cx);
 		let _ = self.protocol_controller_future.poll_unpin(cx);
-
 		self.inner.poll(cx, params)
 	}
 }

substrate/client/network/src/protocol/notifications/upgrade/collec.rs

@@ -17,7 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use futures::prelude::*;
-use libp2p::core::upgrade::{InboundUpgrade, UpgradeInfo};
+use libp2p::core::upgrade::{InboundUpgrade, ProtocolName, UpgradeInfo};
 use std::{
 	iter::FromIterator,
 	pin::Pin,
@@ -76,9 +76,9 @@ where
 #[derive(Debug, Clone, PartialEq)]
 pub struct ProtoNameWithUsize<T>(T, usize);
 
-impl<T: AsRef<str>> AsRef<str> for ProtoNameWithUsize<T> {
-	fn as_ref(&self) -> &str {
-		self.0.as_ref()
+impl<T: ProtocolName> ProtocolName for ProtoNameWithUsize<T> {
+	fn protocol_name(&self) -> &[u8] {
+		self.0.protocol_name()
 	}
 }
 
@@ -104,13 +104,13 @@ impl<T: Future<Output = Result<O, E>>, O, E> Future for FutWithUsize<T> {
 mod tests {
 	use super::*;
 	use crate::types::ProtocolName as ProtoName;
-	use libp2p::core::upgrade::UpgradeInfo;
+	use libp2p::core::upgrade::{ProtocolName, UpgradeInfo};
 
 	// TODO: move to mocks
 	mockall::mock! {
 		pub ProtocolUpgrade<T> {}
 
-		impl<T: Clone + AsRef<str>> UpgradeInfo for ProtocolUpgrade<T> {
+		impl<T: Clone + ProtocolName> UpgradeInfo for ProtocolUpgrade<T> {
 			type Info = T;
 			type InfoIter = vec::IntoIter<T>;
 			fn protocol_info(&self) -> vec::IntoIter<T>;

substrate/client/network/src/protocol/notifications/upgrade/notifications.rs

@@ -114,6 +114,13 @@ pub struct NotificationsOutSubstream<TSubstream> {
 	socket: Framed<TSubstream, UviBytes<io::Cursor<Vec<u8>>>>,
 }
 
+#[cfg(test)]
+impl<TSubstream> NotificationsOutSubstream<TSubstream> {
+	pub fn new(socket: Framed<TSubstream, UviBytes<io::Cursor<Vec<u8>>>>) -> Self {
+		Self { socket }
+	}
+}
+
 impl NotificationsIn {
 	/// Builds a new potential upgrade.
 	pub fn new(
@@ -196,13 +203,13 @@ impl<TSubstream> NotificationsInSubstream<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite + Unpin,
 {
-	// #[cfg(test)]
-	// pub fn new(
-	// 	socket: Framed<TSubstream, UviBytes<io::Cursor<Vec<u8>>>>,
-	// 	handshake: NotificationsInSubstreamHandshake,
-	// ) -> Self {
-	// 	Self { socket, handshake }
-	// }
+	#[cfg(test)]
+	pub fn new(
+		socket: Framed<TSubstream, UviBytes<io::Cursor<Vec<u8>>>>,
+		handshake: NotificationsInSubstreamHandshake,
+	) -> Self {
+		Self { socket, handshake }
+	}
 
 	/// Sends the handshake in order to inform the remote that we accept the substream.
 	pub fn send_handshake(&mut self, message: impl Into<Vec<u8>>) {
@@ -491,92 +498,41 @@ pub enum NotificationsOutError {
 
 #[cfg(test)]
 mod tests {
-	use super::*;
-	use futures::channel::oneshot;
-	use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
+	use super::{NotificationsIn, NotificationsInOpen, NotificationsOut, NotificationsOutOpen};
+	use futures::{channel::oneshot, prelude::*};
+	use libp2p::core::upgrade;
 	use tokio::net::{TcpListener, TcpStream};
 	use tokio_util::compat::TokioAsyncReadCompatExt;
 
-	/// Opens a substream to the given address, negotiates the protocol, and returns the substream
-	/// along with the handshake message.
-	async fn dial(
-		addr: std::net::SocketAddr,
-		handshake: impl Into<Vec<u8>>,
-	) -> Result<
-		(
-			Vec<u8>,
-			NotificationsOutSubstream<
-				multistream_select::Negotiated<tokio_util::compat::Compat<TcpStream>>,
-			>,
-		),
-		NotificationsHandshakeError,
-	> {
-		let socket = TcpStream::connect(addr).await.unwrap();
-		let notifs_out = NotificationsOut::new("/test/proto/1", Vec::new(), handshake, 1024 * 1024);
-		let (_, substream) = multistream_select::dialer_select_proto(
-			socket.compat(),
-			notifs_out.protocol_info().into_iter(),
-			upgrade::Version::V1,
-		)
-		.await
-		.unwrap();
-		let NotificationsOutOpen { handshake, substream, .. } =
-			<NotificationsOut as OutboundUpgrade<_>>::upgrade_outbound(
-				notifs_out,
-				substream,
-				"/test/proto/1".into(),
-			)
-			.await?;
-		Ok((handshake, substream))
-	}
-
-	/// Listens on a localhost, negotiates the protocol, and returns the substream along with the
-	/// handshake message.
-	///
-	/// Also sends the listener address through the given channel.
-	async fn listen_on_localhost(
-		listener_addr_tx: oneshot::Sender<std::net::SocketAddr>,
-	) -> Result<
-		(
-			Vec<u8>,
-			NotificationsInSubstream<
-				multistream_select::Negotiated<tokio_util::compat::Compat<TcpStream>>,
-			>,
-		),
-		NotificationsHandshakeError,
-	> {
-		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
-
-		let (socket, _) = listener.accept().await.unwrap();
-		let notifs_in = NotificationsIn::new("/test/proto/1", Vec::new(), 1024 * 1024);
-		let (_, substream) =
-			multistream_select::listener_select_proto(socket.compat(), notifs_in.protocol_info())
-				.await
-				.unwrap();
-		let NotificationsInOpen { handshake, substream, .. } =
-			<NotificationsIn as InboundUpgrade<_>>::upgrade_inbound(
-				notifs_in,
-				substream,
-				"/test/proto/1".into(),
-			)
-			.await?;
-		Ok((handshake, substream))
-	}
-
 	#[tokio::test]
 	async fn basic_works() {
+		const PROTO_NAME: &str = "/test/proto/1";
 		let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
 
 		let client = tokio::spawn(async move {
-			let (handshake, mut substream) =
-				dial(listener_addr_rx.await.unwrap(), &b"initial message"[..]).await.unwrap();
+			let socket = TcpStream::connect(listener_addr_rx.await.unwrap()).await.unwrap();
+			let NotificationsOutOpen { handshake, mut substream, .. } = upgrade::apply_outbound(
+				socket.compat(),
+				NotificationsOut::new(PROTO_NAME, Vec::new(), &b"initial message"[..], 1024 * 1024),
+				upgrade::Version::V1,
+			)
+			.await
+			.unwrap();
 
 			assert_eq!(handshake, b"hello world");
 			substream.send(b"test message".to_vec()).await.unwrap();
 		});
 
-		let (handshake, mut substream) = listen_on_localhost(listener_addr_tx).await.unwrap();
+		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
+
+		let (socket, _) = listener.accept().await.unwrap();
+		let NotificationsInOpen { handshake, mut substream, .. } = upgrade::apply_inbound(
+			socket.compat(),
+			NotificationsIn::new(PROTO_NAME, Vec::new(), 1024 * 1024),
+		)
+		.await
+		.unwrap();
 
 		assert_eq!(handshake, b"initial message");
 		substream.send_handshake(&b"hello world"[..]);
@@ -591,17 +547,33 @@ mod tests {
 	async fn empty_handshake() {
 		// Check that everything still works when the handshake messages are empty.
 
+		const PROTO_NAME: &str = "/test/proto/1";
 		let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
 
 		let client = tokio::spawn(async move {
-			let (handshake, mut substream) =
-				dial(listener_addr_rx.await.unwrap(), vec![]).await.unwrap();
+			let socket = TcpStream::connect(listener_addr_rx.await.unwrap()).await.unwrap();
+			let NotificationsOutOpen { handshake, mut substream, .. } = upgrade::apply_outbound(
+				socket.compat(),
+				NotificationsOut::new(PROTO_NAME, Vec::new(), vec![], 1024 * 1024),
+				upgrade::Version::V1,
+			)
+			.await
+			.unwrap();
 
 			assert!(handshake.is_empty());
 			substream.send(Default::default()).await.unwrap();
 		});
 
-		let (handshake, mut substream) = listen_on_localhost(listener_addr_tx).await.unwrap();
+		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
+
+		let (socket, _) = listener.accept().await.unwrap();
+		let NotificationsInOpen { handshake, mut substream, .. } = upgrade::apply_inbound(
+			socket.compat(),
+			NotificationsIn::new(PROTO_NAME, Vec::new(), 1024 * 1024),
+		)
+		.await
+		.unwrap();
 
 		assert!(handshake.is_empty());
 		substream.send_handshake(vec![]);
@@ -614,10 +586,17 @@ mod tests {
 
 	#[tokio::test]
 	async fn refused() {
+		const PROTO_NAME: &str = "/test/proto/1";
 		let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
 
 		let client = tokio::spawn(async move {
-			let outcome = dial(listener_addr_rx.await.unwrap(), &b"hello"[..]).await;
+			let socket = TcpStream::connect(listener_addr_rx.await.unwrap()).await.unwrap();
+			let outcome = upgrade::apply_outbound(
+				socket.compat(),
+				NotificationsOut::new(PROTO_NAME, Vec::new(), &b"hello"[..], 1024 * 1024),
+				upgrade::Version::V1,
+			)
+			.await;
 
 			// Despite the protocol negotiation being successfully conducted on the listener
 			// side, we have to receive an error here because the listener didn't send the
@@ -625,7 +604,16 @@ mod tests {
 			assert!(outcome.is_err());
 		});
 
-		let (handshake, substream) = listen_on_localhost(listener_addr_tx).await.unwrap();
+		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
+
+		let (socket, _) = listener.accept().await.unwrap();
+		let NotificationsInOpen { handshake, substream, .. } = upgrade::apply_inbound(
+			socket.compat(),
+			NotificationsIn::new(PROTO_NAME, Vec::new(), 1024 * 1024),
+		)
+		.await
+		.unwrap();
 
 		assert_eq!(handshake, b"hello");
 
@@ -637,16 +625,35 @@ mod tests {
 
 	#[tokio::test]
 	async fn large_initial_message_refused() {
+		const PROTO_NAME: &str = "/test/proto/1";
 		let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
 
 		let client = tokio::spawn(async move {
-			let ret =
-				dial(listener_addr_rx.await.unwrap(), (0..32768).map(|_| 0).collect::<Vec<_>>())
-					.await;
+			let socket = TcpStream::connect(listener_addr_rx.await.unwrap()).await.unwrap();
+			let ret = upgrade::apply_outbound(
+				socket.compat(),
+				// We check that an initial message that is too large gets refused.
+				NotificationsOut::new(
+					PROTO_NAME,
+					Vec::new(),
+					(0..32768).map(|_| 0).collect::<Vec<_>>(),
+					1024 * 1024,
+				),
+				upgrade::Version::V1,
+			)
+			.await;
 			assert!(ret.is_err());
 		});
 
-		let ret = listen_on_localhost(listener_addr_tx).await;
+		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
+
+		let (socket, _) = listener.accept().await.unwrap();
+		let ret = upgrade::apply_inbound(
+			socket.compat(),
+			NotificationsIn::new(PROTO_NAME, Vec::new(), 1024 * 1024),
+		)
+		.await;
 		assert!(ret.is_err());
 
 		client.await.unwrap();
@@ -654,14 +661,30 @@ mod tests {
 
 	#[tokio::test]
 	async fn large_handshake_refused() {
+		const PROTO_NAME: &str = "/test/proto/1";
 		let (listener_addr_tx, listener_addr_rx) = oneshot::channel();
 
 		let client = tokio::spawn(async move {
-			let ret = dial(listener_addr_rx.await.unwrap(), &b"initial message"[..]).await;
+			let socket = TcpStream::connect(listener_addr_rx.await.unwrap()).await.unwrap();
+			let ret = upgrade::apply_outbound(
+				socket.compat(),
+				NotificationsOut::new(PROTO_NAME, Vec::new(), &b"initial message"[..], 1024 * 1024),
+				upgrade::Version::V1,
+			)
+			.await;
 			assert!(ret.is_err());
 		});
 
-		let (handshake, mut substream) = listen_on_localhost(listener_addr_tx).await.unwrap();
+		let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+		listener_addr_tx.send(listener.local_addr().unwrap()).unwrap();
+
+		let (socket, _) = listener.accept().await.unwrap();
+		let NotificationsInOpen { handshake, mut substream, .. } = upgrade::apply_inbound(
+			socket.compat(),
+			NotificationsIn::new(PROTO_NAME, Vec::new(), 1024 * 1024),
+		)
+		.await
+		.unwrap();
 		assert_eq!(handshake, b"initial message");
 
 		// We check that a handshake that is too large gets refused.