Fix clearing queued blocks in the sync module (#11763)

substrate/client/network/sync/src/blocks.rs

@@ -18,7 +18,7 @@
 
 use crate::message;
 use libp2p::PeerId;
-use log::{debug, trace};
+use log::trace;
 use sp_runtime::traits::{Block as BlockT, NumberFor, One};
 use std::{
 	cmp,
@@ -212,18 +212,13 @@ impl<B: BlockT> BlockCollection<B> {
 	}
 
 	pub fn clear_queued(&mut self, from_hash: &B::Hash) {
-		match self.queued_blocks.remove(from_hash) {
+		if let Some((from, to)) = self.queued_blocks.remove(from_hash) {
-			None => {
+			let mut block_num = from;
-				debug!(target: "sync", "Can't clear unknown queued blocks from {:?}", from_hash);
+			while block_num < to {
-			},
+				self.blocks.remove(&block_num);
-			Some((from, to)) => {
+				block_num += One::one();
-				let mut block_num = from;
+			}
-				while block_num < to {
+			trace!(target: "sync", "Cleared blocks from {:?} to {:?}", from, to);
-					self.blocks.remove(&block_num);
-					block_num += One::one();
-				}
-				trace!(target: "sync", "Cleared blocks from {:?} to {:?}", from, to);
-			},
 		}
 	}
 

substrate/client/network/sync/src/lib.rs

@@ -1520,9 +1520,7 @@ where
 		let mut has_error = false;
 		for (_, hash) in &results {
 			self.queue_blocks.remove(hash);
-		}
+			self.blocks.clear_queued(hash);
-		if let Some(from_hash) = results.first().map(|(_, h)| h) {
-			self.blocks.clear_queued(from_hash);
 		}
 		for (result, hash) in results {
 			if has_error {
@@ -3299,23 +3297,34 @@ mod test {
 				);
 			}
 
+			let mut notify_imported: Vec<_> = resp_blocks
+				.iter()
+				.rev()
+				.map(|b| {
+					(
+						Ok(BlockImportStatus::ImportedUnknown(
+							b.header().number().clone(),
+							Default::default(),
+							Some(peer_id1.clone()),
+						)),
+						b.hash(),
+					)
+				})
+				.collect();
+
+			// The import queue may send notifications in batches of varying size. So we simulate
+			// this here by splitting the batch into 2 notifications.
+			let second_batch = notify_imported.split_off(notify_imported.len() / 2);
 			let _ = sync.on_blocks_processed(
 				MAX_BLOCKS_TO_REQUEST as usize,
 				MAX_BLOCKS_TO_REQUEST as usize,
-				resp_blocks
+				notify_imported,
-					.iter()
+			);
-					.rev()
+
-					.map(|b| {
+			let _ = sync.on_blocks_processed(
-						(
+				MAX_BLOCKS_TO_REQUEST as usize,
-							Ok(BlockImportStatus::ImportedUnknown(
+				MAX_BLOCKS_TO_REQUEST as usize,
-								b.header().number().clone(),
+				second_batch,
-								Default::default(),
-								Some(peer_id1.clone()),
-							)),
-							b.hash(),
-						)
-					})
-					.collect(),
 			);
 
 			resp_blocks