Sync: Propagate block announcement data (#7903)

* Sync: Propagate block announcement data

This pr adds a feature to the sync protocol to propagate the data that
we received alongside a block announcement. This is done by adding a
cache that caches the last X block announcement data where X is set to
the number of `in_peers` (giving every peer the chance to send us a
different block). This will be required by parachains to ensure that
even peers who are not connected to a collator receive the data
alongside the block announcement to properly validate it and request the
block.

* Review comment

* Bring back the code and add new variant to ensure we don't insert block
announce data when something wasn't checked

* Also use out_peers

substrate/client/network/src/protocol.rs

@@ -230,6 +230,8 @@ pub struct Protocol<B: BlockT, H: ExHashT> {
 	metrics: Option<Metrics>,
 	/// The `PeerId`'s of all boot nodes.
 	boot_node_ids: HashSet<PeerId>,
+	/// A cache for the data that was associated to a block announcement.
+	block_announce_data_cache: lru::LruCache<B::Hash, Vec<u8>>,
 }
 
 /// Peer information
@@ -491,6 +493,11 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 			)
 		};
 
+		let block_announce_data_cache = lru::LruCache::new(
+			network_config.default_peers_set.in_peers as usize
+				+ network_config.default_peers_set.out_peers as usize,
+		);
+
 		let protocol = Protocol {
 			tick_timeout: Box::pin(interval(TICK_TIMEOUT)),
 			propagate_timeout: Box::pin(interval(PROPAGATE_TIMEOUT)),
@@ -514,6 +521,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 				None
 			},
 			boot_node_ids,
+			block_announce_data_cache,
 		};
 
 		Ok((protocol, peerset_handle, known_addresses))
@@ -1069,7 +1077,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 	///
 	/// In chain-based consensus, we often need to make sure non-best forks are
 	/// at least temporarily synced.
-	pub fn announce_block(&mut self, hash: B::Hash, data: Vec<u8>) {
+	pub fn announce_block(&mut self, hash: B::Hash, data: Option<Vec<u8>>) {
 		let header = match self.chain.header(BlockId::Hash(hash)) {
 			Ok(Some(header)) => header,
 			Ok(None) => {
@@ -1090,6 +1098,8 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		let is_best = self.chain.info().best_hash == hash;
 		debug!(target: "sync", "Reannouncing block {:?} is_best: {}", hash, is_best);
 
+		let data = data.or_else(|| self.block_announce_data_cache.get(&hash).cloned()).unwrap_or_default();
+
 		for (who, ref mut peer) in self.peers.iter_mut() {
 			let inserted = peer.known_blocks.insert(hash);
 			if inserted {
@@ -1160,9 +1170,17 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		validation_result: sync::PollBlockAnnounceValidation<B::Header>,
 	) -> CustomMessageOutcome<B> {
 		let (header, is_best, who) = match validation_result {
-			sync::PollBlockAnnounceValidation::Nothing { is_best, who, header } => {
+			sync::PollBlockAnnounceValidation::Skip =>
+				return CustomMessageOutcome::None,
+			sync::PollBlockAnnounceValidation::Nothing { is_best, who, announce } => {
 				self.update_peer_info(&who);
 
+				if let Some(data) = announce.data {
+					if !data.is_empty() {
+						self.block_announce_data_cache.put(announce.header.hash(), data);
+					}
+				}
+
 				// `on_block_announce` returns `OnBlockAnnounce::ImportHeader`
 				// when we have all data required to import the block
 				// in the BlockAnnounce message. This is only when:
@@ -1170,14 +1188,21 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 				// AND
 				// 2) parent block is already imported and not pruned.
 				if is_best {
-					return CustomMessageOutcome::PeerNewBest(who, *header.number())
+					return CustomMessageOutcome::PeerNewBest(who, *announce.header.number())
 				} else {
 					return CustomMessageOutcome::None
 				}
 			}
-			sync::PollBlockAnnounceValidation::ImportHeader { header, is_best, who } => {
+			sync::PollBlockAnnounceValidation::ImportHeader { announce, is_best, who } => {
 				self.update_peer_info(&who);
-				(header, is_best, who)
+
+				if let Some(data) = announce.data {
+					if !data.is_empty() {
+						self.block_announce_data_cache.put(announce.header.hash(), data);
+					}
+				}
+
+				(announce.header, is_best, who)
 			}
 			sync::PollBlockAnnounceValidation::Failure { who, disconnect } => {
 				if disconnect {

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

@@ -362,8 +362,8 @@ pub enum PollBlockAnnounceValidation<H> {
 		who: PeerId,
 		/// Was this their new best block?
 		is_best: bool,
-		/// The header of the announcement.
-		header: H,
+		/// The announcement.
+		announce: BlockAnnounce<H>,
 	},
 	/// The announcement header should be imported.
 	ImportHeader {
@@ -371,9 +371,11 @@ pub enum PollBlockAnnounceValidation<H> {
 		who: PeerId,
 		/// Was this their new best block?
 		is_best: bool,
-		/// The header of the announcement.
-		header: H,
+		/// The announcement.
+		announce: BlockAnnounce<H>,
 	},
+	/// The block announcement should be skipped.
+	Skip,
 }
 
 /// Result of [`ChainSync::block_announce_validation`].
@@ -388,15 +390,6 @@ enum PreValidateBlockAnnounce<H> {
 		/// Should the peer be disconnected?
 		disconnect: bool,
 	},
-	/// The announcement does not require further handling.
-	Nothing {
-		/// Who sent the processed block announcement?
-		who: PeerId,
-		/// Was this their new best block?
-		is_best: bool,
-		/// The announcement.
-		announce: BlockAnnounce<H>,
-	},
 	/// The pre-validation was sucessful and the announcement should be
 	/// further processed.
 	Process {
@@ -407,6 +400,8 @@ enum PreValidateBlockAnnounce<H> {
 		/// The announcement.
 		announce: BlockAnnounce<H>,
 	},
+	/// The block announcement should be skipped.
+	Skip,
 }
 
 /// Result of [`ChainSync::on_block_justification`].
@@ -1278,7 +1273,7 @@ impl<B: BlockT> ChainSync<B> {
 					who,
 					hash,
 				);
-				PreValidateBlockAnnounce::Nothing { is_best, who, announce }
+				PreValidateBlockAnnounce::Skip
 			}.boxed());
 			return
 		}
@@ -1295,7 +1290,7 @@ impl<B: BlockT> ChainSync<B> {
 						hash,
 						who,
 					);
-					PreValidateBlockAnnounce::Nothing { is_best, who, announce }
+					PreValidateBlockAnnounce::Skip
 				}.boxed());
 				return
 			}
@@ -1308,7 +1303,7 @@ impl<B: BlockT> ChainSync<B> {
 						hash,
 						who,
 					);
-					PreValidateBlockAnnounce::Nothing { is_best, who, announce }
+					PreValidateBlockAnnounce::Skip
 				}.boxed());
 				return
 			}
@@ -1337,7 +1332,7 @@ impl<B: BlockT> ChainSync<B> {
 				}
 				Err(e) => {
 					error!(target: "sync", "💔 Block announcement validation errored: {}", e);
-					PreValidateBlockAnnounce::Nothing { is_best, who, announce }
+					PreValidateBlockAnnounce::Skip
 				}
 			}
 		}.boxed());
@@ -1393,10 +1388,6 @@ impl<B: BlockT> ChainSync<B> {
 		);
 
 		let (announce, is_best, who) = match pre_validation_result {
-			PreValidateBlockAnnounce::Nothing { is_best, who, announce } => {
-				self.peer_block_announce_validation_finished(&who);
-				return PollBlockAnnounceValidation::Nothing { is_best, who, header: announce.header }
-			},
 			PreValidateBlockAnnounce::Failure { who, disconnect } => {
 				self.peer_block_announce_validation_finished(&who);
 				return PollBlockAnnounceValidation::Failure { who, disconnect }
@@ -1405,12 +1396,12 @@ impl<B: BlockT> ChainSync<B> {
 				self.peer_block_announce_validation_finished(&who);
 				(announce, is_new_best, who)
 			},
+			PreValidateBlockAnnounce::Skip => return PollBlockAnnounceValidation::Skip,
 		};
 
-		let header = announce.header;
-		let number = *header.number();
-		let hash = header.hash();
-		let parent_status = self.block_status(header.parent_hash()).unwrap_or(BlockStatus::Unknown);
+		let number = *announce.header.number();
+		let hash = announce.header.hash();
+		let parent_status = self.block_status(announce.header.parent_hash()).unwrap_or(BlockStatus::Unknown);
 		let known_parent = parent_status != BlockStatus::Unknown;
 		let ancient_parent = parent_status == BlockStatus::InChainPruned;
 
@@ -1419,7 +1410,7 @@ impl<B: BlockT> ChainSync<B> {
 			peer
 		} else {
 			error!(target: "sync", "💔 Called on_block_announce with a bad peer ID");
-			return PollBlockAnnounceValidation::Nothing { is_best, who, header }
+			return PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 		};
 
 		if is_best {
@@ -1430,7 +1421,7 @@ impl<B: BlockT> ChainSync<B> {
 
 		if let PeerSyncState::AncestorSearch {..} = peer.state {
 			trace!(target: "sync", "Peer state is ancestor search.");
-			return PollBlockAnnounceValidation::Nothing { is_best, who, header }
+			return PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 		}
 
 		// If the announced block is the best they have and is not ahead of us, our common number
@@ -1438,7 +1429,7 @@ impl<B: BlockT> ChainSync<B> {
 		if is_best {
 			if known && self.best_queued_number >= number {
 				peer.update_common_number(number);
-			} else if header.parent_hash() == &self.best_queued_hash
+			} else if announce.header.parent_hash() == &self.best_queued_hash
 				|| known_parent && self.best_queued_number >= number
 			{
 				peer.update_common_number(number - One::one());
@@ -1452,37 +1443,52 @@ impl<B: BlockT> ChainSync<B> {
 			if let Some(target) = self.fork_targets.get_mut(&hash) {
 				target.peers.insert(who.clone());
 			}
-			return PollBlockAnnounceValidation::Nothing { is_best, who, header }
+			return PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 		}
 
 		if ancient_parent {
-			trace!(target: "sync", "Ignored ancient block announced from {}: {} {:?}", who, hash, header);
-			return PollBlockAnnounceValidation::Nothing { is_best, who, header }
+			trace!(
+				target: "sync",
+				"Ignored ancient block announced from {}: {} {:?}",
+				who,
+				hash,
+				announce.header,
+			);
+			return PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 		}
 
 		let requires_additional_data = !self.role.is_light() || !known_parent;
 		if !requires_additional_data {
-			trace!(target: "sync", "Importing new header announced from {}: {} {:?}", who, hash, header);
-			return PollBlockAnnounceValidation::ImportHeader { is_best, header, who }
+			trace!(
+				target: "sync",
+				"Importing new header announced from {}: {} {:?}",
+				who,
+				hash,
+				announce.header,
+			);
+			return PollBlockAnnounceValidation::ImportHeader { is_best, announce, who }
 		}
 
 		if number <= self.best_queued_number {
 			trace!(
 				target: "sync",
-				"Added sync target for block announced from {}: {} {:?}", who, hash, header
+				"Added sync target for block announced from {}: {} {:?}",
+				who,
+				hash,
+				announce.header,
 			);
 			self.fork_targets
 				.entry(hash.clone())
 				.or_insert_with(|| ForkTarget {
 					number,
-					parent_hash: Some(*header.parent_hash()),
+					parent_hash: Some(*announce.header.parent_hash()),
 					peers: Default::default(),
 				})
 				.peers.insert(who.clone());
 		}
 
 		trace!(target: "sync", "Announce validation result is nothing");
-		PollBlockAnnounceValidation::Nothing { is_best, who, header }
+		PollBlockAnnounceValidation::Nothing { is_best, who, announce }
 	}
 
 	/// Call when a peer has disconnected.

substrate/client/network/src/service.rs

@@ -858,7 +858,7 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
 	///
 	/// In chain-based consensus, we often need to make sure non-best forks are
 	/// at least temporarily synced. This function forces such an announcement.
-	pub fn announce_block(&self, hash: B::Hash, data: Vec<u8>) {
+	pub fn announce_block(&self, hash: B::Hash, data: Option<Vec<u8>>) {
 		let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::AnnounceBlock(hash, data));
 	}
 
@@ -1236,7 +1236,7 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
 	PropagateTransaction(H),
 	PropagateTransactions,
 	RequestJustification(B::Hash, NumberFor<B>),
-	AnnounceBlock(B::Hash, Vec<u8>),
+	AnnounceBlock(B::Hash, Option<Vec<u8>>),
 	GetValue(record::Key),
 	PutValue(record::Key, Vec<u8>),
 	AddKnownAddress(PeerId, Multiaddr),