[DO NOT MERGE] network/libp2p: Investigate requests that don't timeout

substrate/client/network/src/request_responses.rs

@@ -359,6 +359,9 @@ pub struct RequestResponsesBehaviour {
 
 	/// Primarily used to get a reputation of a node.
 	peer_store: Arc<dyn PeerStoreProvider>,
+
+	/// Interval to check for periodic requests.
+	periodic_request_check: tokio::time::Interval,
 }
 
 /// Generated by the response builder and waiting to be processed.
@@ -412,6 +415,7 @@ impl RequestResponsesBehaviour {
 			pending_responses_arrival_time: Default::default(),
 			send_feedback: Default::default(),
 			peer_store,
+			periodic_request_check: tokio::time::interval(Duration::from_secs(60)),
 		})
 	}
 
@@ -659,6 +663,36 @@ impl NetworkBehaviour for RequestResponsesBehaviour {
 		params: &mut impl PollParameters,
 	) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
 		'poll_all: loop {
+			if let Poll::Ready(_) = self.periodic_request_check.poll_tick(cx) {
+				let hanged_requests: Vec<_> = self
+					.pending_requests
+					.iter()
+					.filter_map(|(id, req)| {
+						if req.started_at.elapsed() > Duration::from_secs(60) {
+							Some((id.clone(), req.started_at, req.fallback_request.clone()))
+						} else {
+							None
+						}
+					})
+					.collect();
+
+				if !hanged_requests.is_empty() {
+					log::warn!(
+						target: "sub-libp2p",
+						"Requests hanged for more than 60 seconds: {:?}",
+						hanged_requests
+					);
+
+					// TODO: Force close the requests to avoid possible mem leaks
+					// and resume operation of above subsystems.
+
+					// for (id, req) in hanged_requests {
+					// 	let _ = req.response_tx.send(Err(RequestFailure::Timeout));
+					// 	self.pending_requests.remove(&id);
+					// }
+				}
+			}
+
 			// 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 {