Maintain cache of finalized blocks

client/consensus/common/Cargo.toml

@@ -28,6 +28,7 @@ polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch =
 cumulus-primitives-core = { path = "../../../primitives/core" }
 cumulus-relay-chain-interface = { path = "../../relay-chain-interface" }
 cumulus-client-pov-recovery = { path = "../../pov-recovery" }
+schnellru = "0.2.1"
 
 [dev-dependencies]
 futures-timer = "3.0.2"

client/consensus/common/src/parachain_consensus.rs

@@ -18,6 +18,7 @@ use sc_client_api::{
 	Backend, BlockBackend, BlockImportNotification, BlockchainEvents, Finalizer, UsageProvider,
 };
 use sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
+use schnellru::{ByLength, LruMap};
 use sp_blockchain::Error as ClientError;
 use sp_consensus::{BlockOrigin, BlockStatus};
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
@@ -41,69 +42,130 @@ const LOG_TARGET: &str = "cumulus-consensus";
 const RECOVERY_DELAY: RecoveryDelay =
 	RecoveryDelay { min: Duration::ZERO, max: Duration::from_secs(30) };
 
+fn handle_new_finalized_head<P, Block, B>(
+	parachain: &Arc<P>,
+	finalized_head: Vec<u8>,
+	last_seen_finalized_hashes: &mut LruMap<Block::Hash, ()>,
+) where
+	Block: BlockT,
+	B: Backend<Block>,
+	P: Finalizer<Block, B> + UsageProvider<Block> + BlockchainEvents<Block>,
+{
+	let header = match Block::Header::decode(&mut &finalized_head[..]) {
+		Ok(header) => header,
+		Err(err) => {
+			tracing::debug!(
+				target: LOG_TARGET,
+				error = ?err,
+				"Could not decode parachain header while following finalized heads.",
+			);
+			return
+		},
+	};
+
+	let hash = header.hash();
+
+	last_seen_finalized_hashes.insert(hash, ());
+	// don't finalize the same block multiple times.
+	if parachain.usage_info().chain.finalized_hash != hash {
+		tracing::debug!(
+			target: LOG_TARGET,
+			block_hash = ?hash,
+			"Attempting to finalize header.",
+		);
+		if let Err(e) = parachain.finalize_block(hash, None, true) {
+			match e {
+				ClientError::UnknownBlock(_) => tracing::debug!(
+					target: LOG_TARGET,
+					block_hash = ?hash,
+					"Could not finalize block because it is unknown.",
+				),
+				_ => tracing::warn!(
+					target: LOG_TARGET,
+					error = ?e,
+					block_hash = ?hash,
+					"Failed to finalize block",
+				),
+			}
+		}
+	}
+}
+
 /// Follow the finalized head of the given parachain.
 ///
 /// For every finalized block of the relay chain, it will get the included parachain header
 /// corresponding to `para_id` and will finalize it in the parachain.
 async fn follow_finalized_head<P, Block, B, R>(para_id: ParaId, parachain: Arc<P>, relay_chain: R)
 where
 	Block: BlockT,
-	P: Finalizer<Block, B> + UsageProvider<Block>,
+	P: Finalizer<Block, B> + UsageProvider<Block> + BlockchainEvents<Block>,
 	R: RelayChainInterface + Clone,
 	B: Backend<Block>,
 {
 	let finalized_heads = match finalized_heads(relay_chain, para_id).await {
-		Ok(finalized_heads_stream) => finalized_heads_stream,
+		Ok(finalized_heads_stream) => finalized_heads_stream.fuse(),
 		Err(err) => {
 			tracing::error!(target: LOG_TARGET, error = ?err, "Unable to retrieve finalized heads stream.");
 			return
 		},
 	};
 
+	let mut imported_blocks = parachain.import_notification_stream().fuse();
+
 	pin_mut!(finalized_heads);
 
-	loop {
-		let finalized_head = if let Some(h) = finalized_heads.next().await {
-			h
-		} else {
-			tracing::debug!(target: LOG_TARGET, "Stopping following finalized head.");
-			return
-		};
+	// We use this cache to finalize blocks that are imported late.
+	// For example, a block that has been recovered via PoV-Recovery
+	// on a full node can have several minutes delay. With this cache
+	// we have some "memory" of recently finalized blocks.
+	let mut last_seen_finalized_hashes = LruMap::new(ByLength::new(40));
 
-		let header = match Block::Header::decode(&mut &finalized_head[..]) {
-			Ok(header) => header,
-			Err(err) => {
-				tracing::debug!(
-					target: LOG_TARGET,
-					error = ?err,
-					"Could not decode parachain header while following finalized heads.",
-				);
-				continue
+	loop {
+		select! {
+			fin = finalized_heads.next() => {
+				match fin {
+					Some(finalized_head) =>
+						handle_new_finalized_head(&parachain, finalized_head, &mut last_seen_finalized_hashes),
+					None => {
+						tracing::debug!(target: LOG_TARGET, "Stopping following finalized head.");
+						return
+					}
+				}
 			},
-		};
-
-		let hash = header.hash();
-
-		// don't finalize the same block multiple times.
-		if parachain.usage_info().chain.finalized_hash != hash {
-			tracing::debug!(
-				target: LOG_TARGET,
-				block_hash = ?hash,
-				"Attempting to finalize header.",
-			);
-			if let Err(e) = parachain.finalize_block(hash, None, true) {
-				match e {
-					ClientError::UnknownBlock(_) => tracing::debug!(
-						target: LOG_TARGET,
-						block_hash = ?hash,
-						"Could not finalize block because it is unknown.",
-					),
-					_ => tracing::warn!(
-						target: LOG_TARGET,
-						error = ?e,
-						block_hash = ?hash,
-						"Failed to finalize block",
-					),
+			imported = imported_blocks.next() => {
+				match imported {
+					Some(imported_block) => {
+						// When we see a block import that is already finalized, we immediately finalize it.
+						if last_seen_finalized_hashes.peek(&imported_block.hash).is_some() {
+							tracing::debug!(
+								target: LOG_TARGET,
+								"Setting newly imported block as finalized.",
+							);
+
+							if let Err(e) = parachain.finalize_block(imported_block.hash, None, true) {
+								match e {
+									ClientError::UnknownBlock(_) => tracing::debug!(
+										target: LOG_TARGET,
+										block_hash = ?imported_block.hash,
+										"Could not finalize block because it is unknown.",
+									),
+									_ => tracing::warn!(
+										target: LOG_TARGET,
+										error = ?e,
+										block_hash = ?imported_block.hash,
+										"Failed to finalize block",
+									),
+								}
+							}
+						}
+					},
+					None => {
+						tracing::debug!(
+							target: LOG_TARGET,
+							"Stopping following imported blocks.",
+						);
+						return
+					}
 				}
 			}
 		}
@@ -171,15 +233,15 @@ async fn follow_new_best<P, R, Block, B>(
 	R: RelayChainInterface + Clone,
 	B: Backend<Block>,
 {
-	let new_best_relay_heads = match new_best_heads(relay_chain, para_id).await {
+	let new_best_heads = match new_best_heads(relay_chain, para_id).await {
 		Ok(best_relay_heads_stream) => best_relay_heads_stream.fuse(),
 		Err(err) => {
 			tracing::error!(target: LOG_TARGET, error = ?err, "Unable to retrieve best heads stream.");
 			return
 		},
 	};
 
-	pin_mut!(new_best_relay_heads);
+	pin_mut!(new_best_heads);
 
 	let mut imported_blocks = parachain.import_notification_stream().fuse();
 	// The unset best header of the parachain. Will be `Some(_)` when we have imported a relay chain
@@ -189,10 +251,10 @@ async fn follow_new_best<P, R, Block, B>(
 
 	loop {
 		select! {
-			relay_header = new_best_relay_heads.next() => {
-				match relay_header {
-					Some(relay_header) => handle_new_best_parachain_head(
-						relay_header,
+			h = new_best_heads.next() => {
+				match h {
+					Some(header) => handle_new_best_parachain_head(
+						header,
 						&*parachain,
 						&mut unset_best_header,
 						recovery_chan_tx.as_mut(),

test/service/src/lib.rs

@@ -385,7 +385,7 @@ where
 			para_id,
 			relay_chain_interface,
 			import_queue: import_queue_service,
-			relay_chain_slot_duration: Duration::from_millis(600),
+			relay_chain_slot_duration: Duration::from_secs(6),
 		};
 
 		start_full_node(params)?;

zombienet/tests/0002-pov_recovery.feature

@@ -13,14 +13,14 @@ charlie: is up within 60 seconds
 one: is up within 60 seconds
 two: is up within 60 seconds
 
-# wait 30 blocks and register parachain
-validator-3: reports block height is at least 30 within 250 seconds
+# wait 20 blocks and register parachain
+validator-3: reports block height is at least 20 within 250 seconds
 validator-0: js-script ./register-para.js with "2000" within 240 seconds
 validator-0: parachain 2000 is registered within 300 seconds
 
 # check block production
 bob: reports block height is at least 20 within 600 seconds
 alice: reports block height is at least 20 within 600 seconds
 charlie: reports block height is at least 20 within 600 seconds
-one: reports block height is at least 20 within 600 seconds
-two: reports block height is at least 20 within 600 seconds
+one: reports block height is at least 20 within 800 seconds
+two: reports block height is at least 20 within 800 seconds