Do not request blocks below the common number when syncing (#2045)

This changes `BlockCollection` logic so we don't download block ranges
from peers with which we have these ranges already in sync.

Improves situation with
https://github.com/paritytech/polkadot-sdk/issues/1915.

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

@@ -123,20 +123,36 @@ impl<B: BlockT> BlockCollection<B> {
 		let first_different = common + <NumberFor<B>>::one();
 		let count = (count as u32).into();
 		let (mut range, downloading) = {
+			// Iterate through the ranges in `self.blocks` looking for a range to download
 			let mut downloading_iter = self.blocks.iter().peekable();
 			let mut prev: Option<(&NumberFor<B>, &BlockRangeState<B>)> = None;
 			loop {
 				let next = downloading_iter.next();
 				break match (prev, next) {
+					// If we are already downloading this range, request it from `max_parallel`
+					// peers (`max_parallel = 5` by default).
+					// Do not request already downloading range from peers with common number above
+					// the range start.
 					(Some((start, &BlockRangeState::Downloading { ref len, downloading })), _)
-						if downloading < max_parallel =>
+						if downloading < max_parallel && *start >= first_different =>
 						(*start..*start + *len, downloading),
-					(Some((start, r)), Some((next_start, _))) if *start + r.len() < *next_start =>
-						(*start + r.len()..cmp::min(*next_start, *start + r.len() + count), 0), // gap
-					(Some((start, r)), None) => (*start + r.len()..*start + r.len() + count, 0), /* last range */
-					(None, None) => (first_different..first_different + count, 0),               /* empty */
+					// If there is a gap between ranges requested, download this gap unless the peer
+					// has common number above the gap start
+					(Some((start, r)), Some((next_start, _)))
+						if *start + r.len() < *next_start &&
+							*start + r.len() >= first_different =>
+						(*start + r.len()..cmp::min(*next_start, *start + r.len() + count), 0),
+					// Download `count` blocks after the last range requested unless the peer
+					// has common number above this new range
+					(Some((start, r)), None) if *start + r.len() >= first_different =>
+						(*start + r.len()..*start + r.len() + count, 0),
+					// If there are no ranges currently requested, download `count` blocks after
+					// `common` number
+					(None, None) => (first_different..first_different + count, 0),
+					// If the first range starts above `common + 1`, download the gap at the start
 					(None, Some((start, _))) if *start > first_different =>
-						(first_different..cmp::min(first_different + count, *start), 0), /* gap at the start */
+						(first_different..cmp::min(first_different + count, *start), 0),
+					// Move on to the next range pair
 					_ => {
 						prev = next;
 						continue
@@ -365,10 +381,10 @@ mod test {
 		bc.blocks.insert(114305, BlockRangeState::Complete(blocks));
 
 		let peer0 = PeerId::random();
-		assert_eq!(bc.needed_blocks(peer0, 128, 10000, 000, 1, 200), Some(1..100));
-		assert_eq!(bc.needed_blocks(peer0, 128, 10000, 600, 1, 200), None); // too far ahead
+		assert_eq!(bc.needed_blocks(peer0, 128, 10000, 0, 1, 200), Some(1..100));
+		assert_eq!(bc.needed_blocks(peer0, 128, 10000, 0, 1, 200), None); // too far ahead
 		assert_eq!(
-			bc.needed_blocks(peer0, 128, 10000, 600, 1, 200000),
+			bc.needed_blocks(peer0, 128, 10000, 0, 1, 200000),
 			Some(100 + 128..100 + 128 + 128)
 		);
 	}
@@ -431,4 +447,202 @@ mod test {
 		assert!(bc.blocks.is_empty());
 		assert!(bc.queued_blocks.is_empty());
 	}
+
+	#[test]
+	fn downloaded_range_is_requested_from_max_parallel_peers() {
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		// identical ranges requested from 2 peers
+		let max_parallel = 2;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer2 = PeerId::random();
+		let peer3 = PeerId::random();
+
+		// common for all peers
+		let best = 100;
+		let common = 10;
+
+		assert_eq!(
+			bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
+			Some(11..16)
+		);
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(11..16)
+		);
+		assert_eq!(
+			bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
+			Some(16..21)
+		);
+	}
+	#[test]
+	fn downloaded_range_not_requested_from_peers_with_higher_common_number() {
+		// A peer connects with a common number falling behind our best number
+		// (either a fork or lagging behind).
+		// We request a range from this peer starting at its common number + 1.
+		// Even though we have less than `max_parallel` downloads, we do not request
+		// this range from peers with a common number above the start of this range.
+
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		let max_parallel = 2;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer1_best = 20;
+		let peer1_common = 10;
+
+		// `peer2` has first different above the start of the range downloaded from `peer1`
+		let peer2 = PeerId::random();
+		let peer2_best = 20;
+		let peer2_common = 11; // first_different = 12
+
+		assert_eq!(
+			bc.needed_blocks(peer1, count, peer1_best, peer1_common, max_parallel, max_ahead),
+			Some(11..16),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer2, count, peer2_best, peer2_common, max_parallel, max_ahead),
+			Some(16..21),
+		);
+	}
+
+	#[test]
+	fn gap_above_common_number_requested() {
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		let best = 30;
+		// We need at least 3 ranges requested to have a gap, so to minimize the number of peers
+		// set `max_parallel = 1`
+		let max_parallel = 1;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer2 = PeerId::random();
+		let peer3 = PeerId::random();
+
+		let common = 10;
+		assert_eq!(
+			bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
+			Some(11..16),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(16..21),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
+			Some(21..26),
+		);
+
+		// For some reason there is now a gap at 16..21. We just disconnect `peer2`, but it might
+		// also happen that 16..21 received first and got imported if our best is actually >= 15.
+		bc.clear_peer_download(&peer2);
+
+		// Some peer connects with common number below the gap. The gap is requested from it.
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(16..21),
+		);
+	}
+
+	#[test]
+	fn gap_below_common_number_not_requested() {
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		let best = 30;
+		// We need at least 3 ranges requested to have a gap, so to minimize the number of peers
+		// set `max_parallel = 1`
+		let max_parallel = 1;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer2 = PeerId::random();
+		let peer3 = PeerId::random();
+
+		let common = 10;
+		assert_eq!(
+			bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
+			Some(11..16),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(16..21),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer3, count, best, common, max_parallel, max_ahead),
+			Some(21..26),
+		);
+
+		// For some reason there is now a gap at 16..21. We just disconnect `peer2`, but it might
+		// also happen that 16..21 received first and got imported if our best is actually >= 15.
+		bc.clear_peer_download(&peer2);
+
+		// Some peer connects with common number above the gap. The gap is not requested from it.
+		let common = 23;
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(26..31), // not 16..21
+		);
+	}
+
+	#[test]
+	fn range_at_the_end_above_common_number_requested() {
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		let best = 30;
+		let max_parallel = 1;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer2 = PeerId::random();
+
+		let common = 10;
+		assert_eq!(
+			bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
+			Some(11..16),
+		);
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(16..21),
+		);
+	}
+
+	#[test]
+	fn range_at_the_end_below_common_number_not_requested() {
+		let mut bc = BlockCollection::new();
+		assert!(is_empty(&bc));
+
+		let count = 5;
+		let best = 30;
+		let max_parallel = 1;
+		let max_ahead = 200;
+
+		let peer1 = PeerId::random();
+		let peer2 = PeerId::random();
+
+		let common = 10;
+		assert_eq!(
+			bc.needed_blocks(peer1, count, best, common, max_parallel, max_ahead),
+			Some(11..16),
+		);
+
+		let common = 20;
+		assert_eq!(
+			bc.needed_blocks(peer2, count, best, common, max_parallel, max_ahead),
+			Some(21..26), // not 16..21
+		);
+	}
 }

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

@@ -520,12 +520,15 @@ where
 				Err(BadPeer(who, rep::BLOCKCHAIN_READ_ERROR))
 			},
 			Ok(BlockStatus::KnownBad) => {
-				info!("💔 New peer with known bad best block {} ({}).", best_hash, best_number);
+				info!("💔 New peer {who} with known bad best block {best_hash} ({best_number}).");
 				Err(BadPeer(who, rep::BAD_BLOCK))
 			},
 			Ok(BlockStatus::Unknown) => {
 				if best_number.is_zero() {
-					info!("💔 New peer with unknown genesis hash {} ({}).", best_hash, best_number);
+					info!(
+						"💔 New peer {} with unknown genesis hash {} ({}).",
+						who, best_hash, best_number,
+					);
 					return Err(BadPeer(who, rep::GENESIS_MISMATCH))
 				}
 
@@ -535,7 +538,8 @@ where
 				if self.queue_blocks.len() > MAJOR_SYNC_BLOCKS.into() {
 					debug!(
 						target:LOG_TARGET,
-						"New peer with unknown best hash {} ({}), assuming common block.",
+						"New peer {} with unknown best hash {} ({}), assuming common block.",
+						who,
 						self.best_queued_hash,
 						self.best_queued_number
 					);
@@ -556,7 +560,7 @@ where
 				let (state, req) = if self.best_queued_number.is_zero() {
 					debug!(
 						target:LOG_TARGET,
-						"New peer with best hash {best_hash} ({best_number}).",
+						"New peer {who} with best hash {best_hash} ({best_number}).",
 					);
 
 					(PeerSyncState::Available, None)
@@ -565,7 +569,8 @@ where
 
 					debug!(
 						target:LOG_TARGET,
-						"New peer with unknown best hash {} ({}), searching for common ancestor.",
+						"New peer {} with unknown best hash {} ({}), searching for common ancestor.",
+						who,
 						best_hash,
 						best_number
 					);
@@ -613,7 +618,7 @@ where
 			Ok(BlockStatus::InChainPruned) => {
 				debug!(
 					target: LOG_TARGET,
-					"New peer with known best hash {best_hash} ({best_number}).",
+					"New peer {who} with known best hash {best_hash} ({best_number}).",
 				);
 				self.peers.insert(
 					who,
@@ -848,8 +853,22 @@ where
 								// We've made progress on this chain since the search was started.
 								// Opportunistically set common number to updated number
 								// instead of the one that started the search.
+								trace!(
+									target: LOG_TARGET,
+									"Ancestry search: opportunistically updating peer {} common number from={} => to={}.",
+									*who,
+									peer.common_number,
+									self.best_queued_number,
+								);
 								peer.common_number = self.best_queued_number;
 							} else if peer.common_number < *current {
+								trace!(
+									target: LOG_TARGET,
+									"Ancestry search: updating peer {} common number from={} => to={}.",
+									*who,
+									peer.common_number,
+									*current,
+								);
 								peer.common_number = *current;
 							}
 						}
@@ -1338,6 +1357,13 @@ where
 			// should be kept in that state.
 			if let PeerSyncState::DownloadingJustification(_) = p.state {
 				// We make sure our commmon number is at least something we have.
+				trace!(
+					target: LOG_TARGET,
+					"Keeping peer {} after restart, updating common number from={} => to={} (our best).",
+					peer_id,
+					p.common_number,
+					self.best_queued_number,
+				);
 				p.common_number = self.best_queued_number;
 				self.peers.insert(peer_id, p);
 				return None

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

@@ -245,113 +245,6 @@ fn build_block(client: &mut Arc<TestClient>, at: Option<Hash>, fork: bool) -> Bl
 	block
 }
 
-/// This test is a regression test as observed on a real network.
-///
-/// The node is connected to multiple peers. Both of these peers are having a best block (1)
-/// that is below our best block (3). Now peer 2 announces a fork of block 3 that we will
-/// request from peer 2. After importing the fork, peer 2 and then peer 1 will announce block 4.
-/// But as peer 1 in our view is still at block 1, we will request block 2 (which we already
-/// have) from it. In the meanwhile peer 2 sends us block 4 and 3 and we send another request
-/// for block 2 to peer 2. Peer 1 answers with block 2 and then peer 2. This will need to
-/// succeed, as we have requested block 2 from both peers.
-#[test]
-fn do_not_report_peer_on_block_response_for_block_request() {
-	sp_tracing::try_init_simple();
-
-	let mut client = Arc::new(TestClientBuilder::new().build());
-	let (_chain_sync_network_provider, chain_sync_network_handle) = NetworkServiceProvider::new();
-
-	let mut sync = ChainSync::new(
-		SyncMode::Full,
-		client.clone(),
-		ProtocolName::from("test-block-announce-protocol"),
-		5,
-		64,
-		None,
-		chain_sync_network_handle,
-	)
-	.unwrap();
-
-	let peer_id1 = PeerId::random();
-	let peer_id2 = PeerId::random();
-
-	let mut client2 = client.clone();
-	let mut build_block_at = |at, import| {
-		let mut block_builder = client2.new_block_at(at, Default::default(), false).unwrap();
-		// Make sure we generate a different block as fork
-		block_builder.push_storage_change(vec![1, 2, 3], Some(vec![4, 5, 6])).unwrap();
-
-		let block = block_builder.build().unwrap().block;
-
-		if import {
-			block_on(client2.import(BlockOrigin::Own, block.clone())).unwrap();
-		}
-
-		block
-	};
-
-	let block1 = build_block(&mut client, None, false);
-	let block2 = build_block(&mut client, None, false);
-	let block3 = build_block(&mut client, None, false);
-	let block3_fork = build_block_at(block2.hash(), false);
-
-	// Add two peers which are on block 1.
-	sync.new_peer(peer_id1, block1.hash(), 1).unwrap();
-	sync.new_peer(peer_id2, block1.hash(), 1).unwrap();
-
-	// Tell sync that our best block is 3.
-	sync.update_chain_info(&block3.hash(), 3);
-
-	// There should be no requests.
-	assert!(sync.block_requests().is_empty());
-
-	// Let peer2 announce a fork of block 3
-	send_block_announce(block3_fork.header().clone(), peer_id2, &mut sync);
-
-	// Import and tell sync that we now have the fork.
-	block_on(client.import(BlockOrigin::Own, block3_fork.clone())).unwrap();
-	sync.update_chain_info(&block3_fork.hash(), 3);
-
-	let block4 = build_block_at(block3_fork.hash(), false);
-
-	// Let peer2 announce block 4 and check that sync wants to get the block.
-	send_block_announce(block4.header().clone(), peer_id2, &mut sync);
-
-	let request = get_block_request(&mut sync, FromBlock::Hash(block4.hash()), 2, &peer_id2);
-
-	// Peer1 announces the same block, but as the common block is still `1`, sync will request
-	// block 2 again.
-	send_block_announce(block4.header().clone(), peer_id1, &mut sync);
-
-	let request2 = get_block_request(&mut sync, FromBlock::Number(2), 1, &peer_id1);
-
-	let response = create_block_response(vec![block4.clone(), block3_fork.clone()]);
-	let res = sync.on_block_data(&peer_id2, Some(request), response).unwrap();
-
-	// We should not yet import the blocks, because there is still an open request for fetching
-	// block `2` which blocks the import.
-	assert!(
-		matches!(res, OnBlockData::Import(ImportBlocksAction{ origin: _, blocks }) if blocks.is_empty())
-	);
-
-	let request3 = get_block_request(&mut sync, FromBlock::Number(2), 1, &peer_id2);
-
-	let response = create_block_response(vec![block2.clone()]);
-	let res = sync.on_block_data(&peer_id1, Some(request2), response).unwrap();
-	assert!(matches!(
-		res,
-		OnBlockData::Import(ImportBlocksAction{ origin: _, blocks })
-			if blocks.iter().all(|b| [2, 3, 4].contains(b.header.as_ref().unwrap().number()))
-	));
-
-	let response = create_block_response(vec![block2.clone()]);
-	let res = sync.on_block_data(&peer_id2, Some(request3), response).unwrap();
-	// Nothing to import
-	assert!(
-		matches!(res, OnBlockData::Import(ImportBlocksAction{ origin: _, blocks }) if blocks.is_empty())
-	);
-}
-
 fn unwrap_from_block_number(from: FromBlock<Hash, u64>) -> u64 {
 	if let FromBlock::Number(from) = from {
 		from