collator-protocol: Handle unknown validator heads

Cargo.toml

@@ -1178,7 +1178,7 @@ sc-transaction-pool-api = { path = "substrate/client/transaction-pool/api", defa
 sc-utils = { path = "substrate/client/utils", default-features = false }
 scale-info = { version = "2.11.1", default-features = false }
 schemars = { version = "0.8.13", default-features = false }
-schnellru = { version = "0.2.1" }
+schnellru = { version = "0.2.3" }
 schnorrkel = { version = "0.11.4", default-features = false }
 seccompiler = { version = "0.4.0" }
 secp256k1 = { version = "0.28.0", default-features = false }

polkadot/node/network/collator-protocol/Cargo.toml

@@ -14,6 +14,7 @@ bitvec = { features = ["alloc"], workspace = true }
 futures = { workspace = true }
 futures-timer = { workspace = true }
 gum = { workspace = true, default-features = true }
+schnellru.workspace = true
 
 sp-core = { workspace = true, default-features = true }
 sp-runtime = { workspace = true, default-features = true }

polkadot/node/network/collator-protocol/src/collator_side/mod.rs

@@ -23,6 +23,7 @@ use bitvec::{bitvec, vec::BitVec};
 use futures::{
 	channel::oneshot, future::Fuse, pin_mut, select, stream::FuturesUnordered, FutureExt, StreamExt,
 };
+use schnellru::{ByLength, LruMap};
 use sp_core::Pair;
 
 use polkadot_node_network_protocol::{
@@ -42,7 +43,7 @@ use polkadot_node_subsystem::{
 		CollatorProtocolMessage, NetworkBridgeEvent, NetworkBridgeTxMessage, ParentHeadData,
 		RuntimeApiMessage,
 	},
-	overseer, CollatorProtocolSenderTrait, FromOrchestra, OverseerSignal, PerLeafSpan,
+	overseer, FromOrchestra, OverseerSignal, PerLeafSpan,
 };
 use polkadot_node_subsystem_util::{
 	backing_implicit_view::View as ImplicitView,
@@ -201,6 +202,11 @@ struct PeerData {
 	view: View,
 	/// Network protocol version.
 	version: CollationVersion,
+	/// Unknown heads in the view.
+	///
+	/// This can happen when the validator is faster at importing a block and sending out its
+	/// `View` than the collator is able to import a block.
+	unknown_heads: LruMap<Hash, (), ByLength>,
 }
 
 /// A type wrapping a collation and it's designated core index.
@@ -1199,9 +1205,10 @@ async fn handle_peer_view_change<Context>(
 	peer_id: PeerId,
 	view: View,
 ) {
-	let PeerData { view: current, version } = match state.peer_data.get_mut(&peer_id) {
-		Some(peer_data) => peer_data,
-		None => return,
+	let Some(PeerData { view: current, version, unknown_heads }) =
+		state.peer_data.get_mut(&peer_id)
+	else {
+		return
 	};
 
 	let added: Vec<Hash> = view.difference(&*current).cloned().collect();
@@ -1229,15 +1236,18 @@ async fn handle_peer_view_change<Context>(
 					new_leaf = ?added,
 					"New leaf in peer's view is unknown",
 				);
+
+				unknown_heads.insert(added, ());
+
 				continue
 			},
 		};
 
 		for block_hash in block_hashes {
-			let per_relay_parent = match state.per_relay_parent.get_mut(block_hash) {
-				Some(per_relay_parent) => per_relay_parent,
-				None => continue,
+			let Some(per_relay_parent) = state.per_relay_parent.get_mut(block_hash) else {
+				continue
 			};
+
 			advertise_collation(
 				ctx,
 				*block_hash,
@@ -1283,10 +1293,13 @@ async fn handle_network_msg<Context>(
 					return Ok(())
 				},
 			};
-			state
-				.peer_data
-				.entry(peer_id)
-				.or_insert_with(|| PeerData { view: View::default(), version });
+			state.peer_data.entry(peer_id).or_insert_with(|| PeerData {
+				view: View::default(),
+				version,
+				// Unlikely that the collator is falling 10 blocks behind and if so, it probably is
+				// not able to keep up any way.
+				unknown_heads: LruMap::new(ByLength::new(10)),
+			});
 
 			if let Some(authority_ids) = maybe_authority {
 				gum::trace!(
@@ -1311,7 +1324,7 @@ async fn handle_network_msg<Context>(
 		},
 		OurViewChange(view) => {
 			gum::trace!(target: LOG_TARGET, ?view, "Own view change");
-			handle_our_view_change(ctx.sender(), state, view).await?;
+			handle_our_view_change(ctx, state, view).await?;
 		},
 		PeerMessage(remote, msg) => {
 			handle_incoming_peer_message(ctx, runtime, state, remote, msg).await?;
@@ -1333,21 +1346,19 @@ async fn handle_network_msg<Context>(
 }
 
 /// Handles our view changes.
-async fn handle_our_view_change<Sender>(
-	sender: &mut Sender,
+#[overseer::contextbounds(CollatorProtocol, prefix = crate::overseer)]
+async fn handle_our_view_change<Context>(
+	ctx: &mut Context,
 	state: &mut State,
 	view: OurView,
-) -> Result<()>
-where
-	Sender: CollatorProtocolSenderTrait,
-{
+) -> Result<()> {
 	let current_leaves = state.active_leaves.clone();
 
 	let removed = current_leaves.iter().filter(|(h, _)| !view.contains(h));
 	let added = view.iter().filter(|h| !current_leaves.contains_key(h));
 
 	for leaf in added {
-		let mode = prospective_parachains_mode(sender, *leaf).await?;
+		let mode = prospective_parachains_mode(ctx.sender(), *leaf).await?;
 
 		if let Some(span) = view.span_per_head().get(leaf).cloned() {
 			let per_leaf_span = PerLeafSpan::new(span, "collator-side");
@@ -1360,19 +1371,44 @@ where
 		if mode.is_enabled() {
 			if let Some(ref mut implicit_view) = state.implicit_view {
 				implicit_view
-					.activate_leaf(sender, *leaf)
+					.activate_leaf(ctx.sender(), *leaf)
 					.await
 					.map_err(Error::ImplicitViewFetchError)?;
 
 				let allowed_ancestry = implicit_view
 					.known_allowed_relay_parents_under(leaf, state.collating_on)
 					.unwrap_or_default();
 
+				// Get the peers that already reported us this head, but we didn't knew it at this
+				// point.
+				let peers = state
+					.peer_data
+					.iter_mut()
+					.filter_map(|(id, data)| {
+						data.unknown_heads.remove(leaf).map(|_| (id, data.version))
+					})
+					.collect::<Vec<_>>();
+
 				for block_hash in allowed_ancestry {
-					state
+					let per_relay_parent = state
 						.per_relay_parent
 						.entry(*block_hash)
 						.or_insert_with(|| PerRelayParent::new(mode));
+
+					// Announce relevant collations to these peers.
+					for (peer_id, peer_version) in &peers {
+						advertise_collation(
+							ctx,
+							*block_hash,
+							per_relay_parent,
+							&peer_id,
+							*peer_version,
+							&state.peer_ids,
+							&mut state.advertisement_timeouts,
+							&state.metrics,
+						)
+						.await;
+					}
 				}
 			}
 		}