net/libp2p: Enforce outbound request-response timeout limits

prdoc/pr_7222.prdoc

@@ -0,0 +1,19 @@
+title: Enforce libp2p outbound request-response timeout limits
+
+doc:
+  - audience: Node Dev
+    description: |
+      This PR enforces that outbound requests are finished within the specified protocol timeout.
+      The stable2412 version running libp2p 0.52.4 contains a bug which does not track request timeouts properly:
+      https://github.com/libp2p/rust-libp2p/pull/5429.
+
+      The issue has been detected while submitting libp2p -> litep2p requests in kusama. 
+      This aims to check that pending outbound requests have not timedout.
+      Although the issue has been fixed in libp2p, there might be other cases where this may happen.
+      For example: https://github.com/libp2p/rust-libp2p/pull/5417.
+
+      For more context see: https://github.com/paritytech/polkadot-sdk/issues/7076#issuecomment-2596085096.
+
+crates:
+- name: sc-network
+  bump: patch

substrate/client/network/src/request_responses.rs

@@ -64,6 +64,9 @@ use std::{
 
 pub use libp2p::request_response::{Config, InboundRequestId, OutboundRequestId};
 
+/// Periodically check if requests are taking too long.
+const PERIODIC_REQUEST_CHECK: Duration = Duration::from_secs(2);
+
 /// Possible failures occurring in the context of sending an outbound request and receiving the
 /// response.
 #[derive(Debug, Clone, thiserror::Error)]
@@ -336,16 +339,20 @@ impl<RequestId> From<(ProtocolName, RequestId)> for ProtocolRequestId<RequestId>
 	}
 }
 
+/// Details of a request-response protocol.
+struct ProtocolDetails {
+	behaviour: Behaviour<GenericCodec>,
+	inbound_queue: Option<async_channel::Sender<IncomingRequest>>,
+	request_timeout: Duration,
+}
+
 /// Implementation of `NetworkBehaviour` that provides support for request-response protocols.
 pub struct RequestResponsesBehaviour {
 	/// The multiple sub-protocols, by name.
 	///
 	/// Contains the underlying libp2p request-response [`Behaviour`], plus an optional
 	/// "response builder" used to build responses for incoming requests.
-	protocols: HashMap<
-		ProtocolName,
-		(Behaviour<GenericCodec>, Option<async_channel::Sender<IncomingRequest>>),
-	>,
+	protocols: HashMap<ProtocolName, ProtocolDetails>,
 
 	/// Pending requests, passed down to a request-response [`Behaviour`], awaiting a reply.
 	pending_requests: HashMap<ProtocolRequestId<OutboundRequestId>, PendingRequest>,
@@ -365,6 +372,14 @@ pub struct RequestResponsesBehaviour {
 
 	/// Primarily used to get a reputation of a node.
 	peer_store: Arc<dyn PeerStoreProvider>,
+
+	/// Interval to check that the requests are not taking too long.
+	///
+	/// We had issues in the past where libp2p did not produce a timeout event in due time.
+	///
+	/// For more details, see:
+	/// - https://github.com/paritytech/polkadot-sdk/issues/7076#issuecomment-2596085096
+	periodic_request_check: tokio::time::Interval,
 }
 
 /// Generated by the response builder and waiting to be processed.
@@ -393,7 +408,7 @@ impl RequestResponsesBehaviour {
 				ProtocolSupport::Outbound
 			};
 
-			let rq_rp = Behaviour::with_codec(
+			let behaviour = Behaviour::with_codec(
 				GenericCodec {
 					max_request_size: protocol.max_request_size,
 					max_response_size: protocol.max_response_size,
@@ -405,7 +420,11 @@ impl RequestResponsesBehaviour {
 			);
 
 			match protocols.entry(protocol.name) {
-				Entry::Vacant(e) => e.insert((rq_rp, protocol.inbound_queue)),
+				Entry::Vacant(e) => e.insert(ProtocolDetails {
+					behaviour,
+					inbound_queue: protocol.inbound_queue,
+					request_timeout: protocol.request_timeout,
+				}),
 				Entry::Occupied(e) => return Err(RegisterError::DuplicateProtocol(e.key().clone())),
 			};
 		}
@@ -417,6 +436,7 @@ impl RequestResponsesBehaviour {
 			pending_responses_arrival_time: Default::default(),
 			send_feedback: Default::default(),
 			peer_store,
+			periodic_request_check: tokio::time::interval(PERIODIC_REQUEST_CHECK),
 		})
 	}
 
@@ -437,9 +457,11 @@ impl RequestResponsesBehaviour {
 	) {
 		log::trace!(target: "sub-libp2p", "send request to {target} ({protocol_name:?}), {} bytes", request.len());
 
-		if let Some((protocol, _)) = self.protocols.get_mut(protocol_name.deref()) {
+		if let Some(ProtocolDetails { behaviour, .. }) =
+			self.protocols.get_mut(protocol_name.deref())
+		{
 			Self::send_request_inner(
-				protocol,
+				behaviour,
 				&mut self.pending_requests,
 				target,
 				protocol_name,
@@ -521,18 +543,19 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 		local_addr: &Multiaddr,
 		remote_addr: &Multiaddr,
 	) -> Result<THandler<Self>, ConnectionDenied> {
-		let iter = self.protocols.iter_mut().filter_map(|(p, (r, _))| {
-			if let Ok(handler) = r.handle_established_inbound_connection(
-				connection_id,
-				peer,
-				local_addr,
-				remote_addr,
-			) {
-				Some((p.to_string(), handler))
-			} else {
-				None
-			}
-		});
+		let iter =
+			self.protocols.iter_mut().filter_map(|(p, ProtocolDetails { behaviour, .. })| {
+				if let Ok(handler) = behaviour.handle_established_inbound_connection(
+					connection_id,
+					peer,
+					local_addr,
+					remote_addr,
+				) {
+					Some((p.to_string(), handler))
+				} else {
+					None
+				}
+			});
 
 		Ok(MultiHandler::try_from_iter(iter).expect(
 			"Protocols are in a HashMap and there can be at most one handler per protocol name, \
@@ -548,19 +571,20 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 		role_override: Endpoint,
 		port_use: PortUse,
 	) -> Result<THandler<Self>, ConnectionDenied> {
-		let iter = self.protocols.iter_mut().filter_map(|(p, (r, _))| {
-			if let Ok(handler) = r.handle_established_outbound_connection(
-				connection_id,
-				peer,
-				addr,
-				role_override,
-				port_use,
-			) {
-				Some((p.to_string(), handler))
-			} else {
-				None
-			}
-		});
+		let iter =
+			self.protocols.iter_mut().filter_map(|(p, ProtocolDetails { behaviour, .. })| {
+				if let Ok(handler) = behaviour.handle_established_outbound_connection(
+					connection_id,
+					peer,
+					addr,
+					role_override,
+					port_use,
+				) {
+					Some((p.to_string(), handler))
+				} else {
+					None
+				}
+			});
 
 		Ok(MultiHandler::try_from_iter(iter).expect(
 			"Protocols are in a HashMap and there can be at most one handler per protocol name, \
@@ -569,8 +593,8 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 	}
 
 	fn on_swarm_event(&mut self, event: FromSwarm) {
-		for (protocol, _) in self.protocols.values_mut() {
-			protocol.on_swarm_event(event);
+		for ProtocolDetails { behaviour, .. } in self.protocols.values_mut() {
+			behaviour.on_swarm_event(event);
 		}
 	}
 
@@ -581,8 +605,8 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 		event: THandlerOutEvent<Self>,
 	) {
 		let p_name = event.0;
-		if let Some((proto, _)) = self.protocols.get_mut(p_name.as_str()) {
-			return proto.on_connection_handler_event(peer_id, connection_id, event.1)
+		if let Some(ProtocolDetails { behaviour, .. }) = self.protocols.get_mut(p_name.as_str()) {
+			return behaviour.on_connection_handler_event(peer_id, connection_id, event.1)
 		} else {
 			log::warn!(
 				target: "sub-libp2p",
@@ -594,6 +618,60 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 
 	fn poll(&mut self, cx: &mut Context) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
 		'poll_all: loop {
+			// Poll the periodic request check.
+			if let Poll::Ready(_) = self.periodic_request_check.poll_tick(cx) {
+				// First pass is needed to collect timed out requests and
+				// submit `OutboundFailure::Timeout` on the moved oneshot sender.
+				let timedout_requests: Vec<_> = self
+					.pending_requests
+					.iter()
+					.filter_map(|(id, req)| {
+						let Some(ProtocolDetails { request_timeout, .. }) =
+							self.protocols.get(&id.protocol)
+						else {
+							log::warn!(
+								target: "sub-libp2p",
+								"Request with id {:?} has no protocol registered.",
+								id.request_id,
+							);
+							return Some(id.clone())
+						};
+
+						let elapsed = req.started_at.elapsed();
+						if elapsed > *request_timeout {
+							Some(id.clone())
+						} else {
+							None
+						}
+					})
+					.collect();
+
+				if !timedout_requests.is_empty() {
+					log::warn!(
+						target: "sub-libp2p",
+						"Requests {timedout_requests:?} force detected as timeout.",
+					);
+				}
+
+				for id in timedout_requests {
+					let Some(req) = self.pending_requests.remove(&id) else {
+						continue;
+					};
+
+					if req
+						.response_tx
+						.send(Err(RequestFailure::Network(OutboundFailure::Timeout)))
+						.is_err()
+					{
+						log::debug!(
+							target: "sub-libp2p",
+							"Request with id {id:?} force detected as timeout. At the same time local \
+							 node is no longer interested in the result.",
+						);
+					}
+				}
+			}
+
 			// Poll to see if any response is ready to be sent back.
 			while let Poll::Ready(Some(outcome)) = self.pending_responses.poll_next_unpin(cx) {
 				let RequestProcessingOutcome {
@@ -610,10 +688,12 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 				};
 
 				if let Ok(payload) = result {
-					if let Some((protocol, _)) = self.protocols.get_mut(&*protocol_name) {
+					if let Some(ProtocolDetails { behaviour, .. }) =
+						self.protocols.get_mut(&*protocol_name)
+					{
 						log::trace!(target: "sub-libp2p", "send response to {peer} ({protocol_name:?}), {} bytes", payload.len());
 
-						if protocol.send_response(inner_channel, Ok(payload)).is_err() {
+						if behaviour.send_response(inner_channel, Ok(payload)).is_err() {
 							// Note: Failure is handled further below when receiving
 							// `InboundFailure` event from request-response [`Behaviour`].
 							log::debug!(
@@ -641,7 +721,8 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 			let mut fallback_requests = vec![];
 
 			// Poll request-responses protocols.
-			for (protocol, (ref mut behaviour, ref mut resp_builder)) in &mut self.protocols {
+			for (protocol, ProtocolDetails { behaviour, inbound_queue, .. }) in &mut self.protocols
+			{
 				'poll_protocol: while let Poll::Ready(ev) = behaviour.poll(cx) {
 					let ev = match ev {
 						// Main events we are interested in.
@@ -696,7 +777,7 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 
 							// Submit the request to the "response builder" passed by the user at
 							// initialization.
-							if let Some(resp_builder) = resp_builder {
+							if let Some(resp_builder) = inbound_queue {
 								// If the response builder is too busy, silently drop `tx`. This
 								// will be reported by the corresponding request-response
 								// [`Behaviour`] through an `InboundFailure::Omission` event.
@@ -904,7 +985,7 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 
 			// Send out fallback requests.
 			for (peer, protocol, request, pending_response) in fallback_requests.drain(..) {
-				if let Some((behaviour, _)) = self.protocols.get_mut(&protocol) {
+				if let Some(ProtocolDetails { behaviour, .. }) = self.protocols.get_mut(&protocol) {
 					Self::send_request_inner(
 						behaviour,
 						&mut self.pending_requests,