Inform sync explicitly about new best block (paritytech#7604)

* Inform sync explicitly about new best block

Instead of "fishing" the new best block out of the processed blocks, we
now tell sync directly that there is a new best block. It also makes
sure that we update the corresponding sync handshake to the new best
block. This is required for parachains as they first import blocks and
declare the new best block after being made aware of it by the relay chain.

* Adds test

* Make sure async stuff had time to run

Author: bkchr

client/network/src/protocol.rs

@@ -528,10 +528,9 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		self.sync.num_sync_requests()
 	}
 
-	/// Sync local state with the blockchain state.
-	pub fn update_chain(&mut self) {
-		let info = self.context_data.chain.info();
-		self.sync.update_chain_info(&info.best_hash, info.best_number);
+	/// Inform sync about new best imported block.
+	pub fn new_best_block_imported(&mut self, hash: B::Hash, number: NumberFor<B>) {
+		self.sync.update_chain_info(&hash, number);
 		self.behaviour.set_legacy_handshake_message(
 			build_status_message(&self.config, &self.context_data.chain),
 		);
@@ -541,11 +540,6 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		);
 	}
 
-	/// Inform sync about an own imported block.
-	pub fn own_block_imported(&mut self, hash: B::Hash, number: NumberFor<B>) {
-		self.sync.update_chain_info(&hash, number);
-	}
-
 	fn update_peer_info(&mut self, who: &PeerId) {
 		if let Some(info) = self.sync.peer_info(who) {
 			if let Some(ref mut peer) = self.context_data.peers.get_mut(who) {
@@ -1258,18 +1252,6 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		count: usize,
 		results: Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>
 	) {
-		let new_best = results.iter().rev().find_map(|r| match r {
-			(Ok(BlockImportResult::ImportedUnknown(n, aux, _)), hash) if aux.is_new_best => Some((*n, hash.clone())),
-			_ => None,
-		});
-		if let Some((best_num, best_hash)) = new_best {
-			self.sync.update_chain_info(&best_hash, best_num);
-			self.behaviour.set_legacy_handshake_message(build_status_message(&self.config, &self.context_data.chain));
-			self.behaviour.set_notif_protocol_handshake(
-				&self.block_announces_protocol,
-				BlockAnnouncesHandshake::build(&self.config, &self.context_data.chain).encode()
-			);
-		}
 		let results = self.sync.on_blocks_processed(
 			imported,
 			count,

client/network/src/service.rs

@@ -484,11 +484,9 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 		self.network_service.user_protocol_mut().on_block_finalized(hash, &header);
 	}
 
-	/// This should be called when blocks are added to the
-	/// chain by something other than the import queue.
-	/// Currently this is only useful for tests.
-	pub fn update_chain(&mut self) {
-		self.network_service.user_protocol_mut().update_chain();
+	/// Inform the network service about new best imported block.
+	pub fn new_best_block_imported(&mut self, hash: B::Hash, number: NumberFor<B>) {
+		self.network_service.user_protocol_mut().new_best_block_imported(hash, number);
 	}
 
 	/// Returns the local `PeerId`.
@@ -1012,21 +1010,11 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
 		self.num_connected.load(Ordering::Relaxed)
 	}
 
-	/// This function should be called when blocks are added to the chain by something other
-	/// than the import queue.
-	///
-	/// > **Important**: This function is a hack and can be removed at any time. Do **not** use it.
-	pub fn update_chain(&self) {
-		let _ = self
-			.to_worker
-			.unbounded_send(ServiceToWorkerMsg::UpdateChain);
-	}
-
-	/// Inform the network service about an own imported block.
-	pub fn own_block_imported(&self, hash: B::Hash, number: NumberFor<B>) {
+	/// Inform the network service about new best imported block.
+	pub fn new_best_block_imported(&self, hash: B::Hash, number: NumberFor<B>) {
 		let _ = self
 			.to_worker
-			.unbounded_send(ServiceToWorkerMsg::OwnBlockImported(hash, number));
+			.unbounded_send(ServiceToWorkerMsg::NewBestBlockImported(hash, number));
 	}
 
 	/// Utility function to extract `PeerId` from each `Multiaddr` for priority group updates.
@@ -1181,8 +1169,7 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
 		protocol_name: Cow<'static, str>,
 	},
 	DisconnectPeer(PeerId),
-	UpdateChain,
-	OwnBlockImported(B::Hash, NumberFor<B>),
+	NewBestBlockImported(B::Hash, NumberFor<B>),
 }
 
 /// Main network worker. Must be polled in order for the network to advance.
@@ -1319,10 +1306,8 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 					this.network_service.register_notifications_protocol(protocol_name),
 				ServiceToWorkerMsg::DisconnectPeer(who) =>
 					this.network_service.user_protocol_mut().disconnect_peer(&who),
-				ServiceToWorkerMsg::UpdateChain =>
-					this.network_service.user_protocol_mut().update_chain(),
-				ServiceToWorkerMsg::OwnBlockImported(hash, number) =>
-					this.network_service.user_protocol_mut().own_block_imported(hash, number),
+				ServiceToWorkerMsg::NewBestBlockImported(hash, number) =>
+					this.network_service.user_protocol_mut().new_best_block_imported(hash, number),
 			}
 		}
 

client/network/test/src/lib.rs

@@ -279,7 +279,7 @@ impl<D> Peer<D> {
 		where F: FnMut(BlockBuilder<Block, PeersFullClient, substrate_test_runtime_client::Backend>) -> Block
 	{
 		let best_hash = self.client.info().best_hash;
-		self.generate_blocks_at(BlockId::Hash(best_hash), count, origin, edit_block, false)
+		self.generate_blocks_at(BlockId::Hash(best_hash), count, origin, edit_block, false, true)
 	}
 
 	/// Add blocks to the peer -- edit the block before adding. The chain will
@@ -291,6 +291,7 @@ impl<D> Peer<D> {
 		origin: BlockOrigin,
 		mut edit_block: F,
 		headers_only: bool,
+		inform_sync_about_new_best_block: bool,
 	) -> H256 where F: FnMut(BlockBuilder<Block, PeersFullClient, substrate_test_runtime_client::Backend>) -> Block {
 		let full_client = self.client.as_full()
 			.expect("blocks could only be generated by full clients");
@@ -328,7 +329,12 @@ impl<D> Peer<D> {
 			at = hash;
 		}
 
-		self.network.update_chain();
+		if inform_sync_about_new_best_block {
+			self.network.new_best_block_imported(
+				at,
+				full_client.header(&BlockId::Hash(at)).ok().flatten().unwrap().number().clone(),
+			);
+		}
 		self.network.service().announce_block(at.clone(), Vec::new());
 		at
 	}
@@ -342,18 +348,36 @@ impl<D> Peer<D> {
 	/// Push blocks to the peer (simplified: with or without a TX)
 	pub fn push_headers(&mut self, count: usize) -> H256 {
 		let best_hash = self.client.info().best_hash;
-		self.generate_tx_blocks_at(BlockId::Hash(best_hash), count, false, true)
+		self.generate_tx_blocks_at(BlockId::Hash(best_hash), count, false, true, true)
 	}
 
 	/// Push blocks to the peer (simplified: with or without a TX) starting from
 	/// given hash.
 	pub fn push_blocks_at(&mut self, at: BlockId<Block>, count: usize, with_tx: bool) -> H256 {
-		self.generate_tx_blocks_at(at, count, with_tx, false)
+		self.generate_tx_blocks_at(at, count, with_tx, false, true)
+	}
+
+	/// Push blocks to the peer (simplified: with or without a TX) starting from
+	/// given hash without informing the sync protocol about the new best block.
+	pub fn push_blocks_at_without_informing_sync(
+		&mut self,
+		at: BlockId<Block>,
+		count: usize,
+		with_tx: bool,
+	) -> H256 {
+		self.generate_tx_blocks_at(at, count, with_tx, false, false)
 	}
 
 	/// Push blocks/headers to the peer (simplified: with or without a TX) starting from
 	/// given hash.
-	fn generate_tx_blocks_at(&mut self, at: BlockId<Block>, count: usize, with_tx: bool, headers_only:bool) -> H256 {
+	fn generate_tx_blocks_at(
+		&mut self,
+		at: BlockId<Block>,
+		count: usize,
+		with_tx: bool,
+		headers_only: bool,
+		inform_sync_about_new_best_block: bool,
+	) -> H256 {
 		let mut nonce = 0;
 		if with_tx {
 			self.generate_blocks_at(
@@ -370,7 +394,8 @@ impl<D> Peer<D> {
 					nonce = nonce + 1;
 					builder.build().unwrap().block
 				},
-				headers_only
+				headers_only,
+				inform_sync_about_new_best_block,
 			)
 		} else {
 			self.generate_blocks_at(
@@ -379,6 +404,7 @@ impl<D> Peer<D> {
 				BlockOrigin::File,
 				|builder| builder.build().unwrap().block,
 				headers_only,
+				inform_sync_about_new_best_block,
 			)
 		}
 	}

client/network/test/src/sync.rs

@@ -779,3 +779,38 @@ fn wait_until_deferred_block_announce_validation_is_ready() {
 		net.block_until_idle();
 	}
 }
+
+/// When we don't inform the sync protocol about the best block, a node will not sync from us as the
+/// handshake is not does not contain our best block.
+#[test]
+fn sync_to_tip_requires_that_sync_protocol_is_informed_about_best_block() {
+	sp_tracing::try_init_simple();
+	log::trace!(target: "sync", "Test");
+	let mut net = TestNet::new(1);
+
+	// Produce some blocks
+	let block_hash = net.peer(0).push_blocks_at_without_informing_sync(BlockId::Number(0), 3, true);
+
+	// Add a node and wait until they are connected
+	net.add_full_peer_with_config(Default::default());
+	net.block_until_connected();
+	net.block_until_idle();
+
+	// The peer should not have synced the block.
+	assert!(!net.peer(1).has_block(&block_hash));
+
+	// Make sync protocol aware of the best block
+	net.peer(0).network_service().new_best_block_imported(block_hash, 3);
+	net.block_until_idle();
+
+	// Connect another node that should now sync to the tip
+	net.add_full_peer_with_config(Default::default());
+	net.block_until_connected();
+
+	while !net.peer(2).has_block(&block_hash) {
+		net.block_until_idle();
+	}
+
+	// However peer 1 should still not have the block.
+	assert!(!net.peer(1).has_block(&block_hash));
+}

client/service/src/lib.rs

@@ -250,8 +250,8 @@ async fn build_network_future<
 					network.service().announce_block(notification.hash, Vec::new());
 				}
 
-				if let sp_consensus::BlockOrigin::Own = notification.origin {
-					network.service().own_block_imported(
+				if notification.is_new_best {
+					network.service().new_best_block_imported(
 						notification.hash,
 						notification.header.number().clone(),
 					);

client/service/test/src/lib.rs

@@ -542,7 +542,8 @@ pub fn sync<G, E, Fb, F, Lb, L, B, ExF, U>(
 
 			make_block_and_import(&first_service, first_user_data);
 		}
-		network.full_nodes[0].1.network().update_chain();
+		let info = network.full_nodes[0].1.client().info();
+		network.full_nodes[0].1.network().new_best_block_imported(info.best_hash, info.best_number);
 		network.full_nodes[0].3.clone()
 	};
 

primitives/consensus/common/src/import_queue.rs

@@ -136,7 +136,7 @@ pub trait Link<B: BlockT>: Send {
 
 /// Block import successful result.
 #[derive(Debug, PartialEq)]
-pub enum BlockImportResult<N: ::std::fmt::Debug + PartialEq> {
+pub enum BlockImportResult<N: std::fmt::Debug + PartialEq> {
 	/// Imported known block.
 	ImportedKnown(N),
 	/// Imported unknown block.