Avoid unnecessary state reset of allowed_requests when no block req…

…uests are sent (#5774)

This PR is cherry-picked from
#5663 so that I can
maintain a smaller polkadot-sdk diff downstream sooner than later.

cc @lexnv @dmitry-markin

---------

Co-authored-by: Alexandru Vasile <[email protected]>
Co-authored-by: Dmitry Markin <[email protected]>

prdoc/pr_5774.prdoc

@@ -0,0 +1,13 @@
+title: Avoid unnecessary state reset of allowed_requests when no block requests are sent
+
+doc:
+  - audience: Node Dev
+    description: |
+      Previously, the state of `allowed_requests` was always reset to the default
+      even if there were no new block requests. This could cause an edge case
+      because `peer_block_request()` will return early next time when there are no ongoing block requests.
+      This patch fixes it by checking whether block requests are empty before updating the state.
+
+crates:
+  - name: sc-network-sync
+    bump: patch

substrate/client/network/sync/src/strategy/chain_sync.rs

@@ -148,6 +148,7 @@ impl Metrics {
 	}
 }
 
+#[derive(Debug, Clone)]
 enum AllowedRequests {
 	Some(HashSet<PeerId>),
 	All,
@@ -1714,13 +1715,14 @@ where
 		let best_queued = self.best_queued_number;
 		let client = &self.client;
 		let queue_blocks = &self.queue_blocks;
-		let allowed_requests = self.allowed_requests.take();
+		let allowed_requests = self.allowed_requests.clone();
 		let max_parallel = if is_major_syncing { 1 } else { self.max_parallel_downloads };
 		let max_blocks_per_request = self.max_blocks_per_request;
 		let gap_sync = &mut self.gap_sync;
 		let disconnected_peers = &mut self.disconnected_peers;
 		let metrics = self.metrics.as_ref();
-		self.peers
+		let requests = self
+			.peers
 			.iter_mut()
 			.filter_map(move |(&id, peer)| {
 				if !peer.state.is_available() ||
@@ -1819,7 +1821,15 @@ where
 					None
 				}
 			})
-			.collect()
+			.collect::<Vec<_>>();
+
+		// Clear the allowed_requests state when sending new block requests
+		// to prevent multiple inflight block requests from being issued.
+		if !requests.is_empty() {
+			self.allowed_requests.take();
+		}
+
+		requests
 	}
 
 	/// Get a state request scheduled by sync to be sent out (if any).