Fix missing block number issue on forced canonicalization (paritytech…

…#12949)

* Fix missing block number issue on forced canonicalization

There is this issue about missing block numbers on forced canonicalization. I looked over the code
now 10000 times and there are possible ways this can be triggered, but I don't really know how this
is triggered. So, this pr is going to solve the symptom and not the cause. The block number to hash
mapping is set when we import a new best block. Forced canonicalization will now stop at the best
block and it will canonicalize the other blocks later when the best block moved. As the error
reports indicated that this issue mainly happened on major sync, there should not be any forks, so
not doing the canonicalization directly shouldn't be that harmful. All known implementations should
import all blocks as best block on major sync anyway (I mean somewhere there is the bug, but I
didn't yet found it).

I will also do some changes to Cumulus around some potential culprit for this issue.

Closes: paritytech#12613

* Add some docs

* Fix fix

* Review comments

* Review comments

client/db/src/lib.rs

@@ -1213,9 +1213,8 @@ impl<Block: BlockT> Backend<Block> {
 
 		let meta = self.blockchain.meta.read();
 
-		if meta.best_number > best_number &&
-			(meta.best_number - best_number).saturated_into::<u64>() >
-				self.canonicalization_delay
+		if meta.best_number.saturating_sub(best_number).saturated_into::<u64>() >
+			self.canonicalization_delay
 		{
 			return Err(sp_blockchain::Error::SetHeadTooOld)
 		}
@@ -1316,42 +1315,43 @@ impl<Block: BlockT> Backend<Block> {
 	fn force_delayed_canonicalize(
 		&self,
 		transaction: &mut Transaction<DbHash>,
-		hash: Block::Hash,
-		number: NumberFor<Block>,
 	) -> ClientResult<()> {
-		let number_u64 = number.saturated_into::<u64>();
-		if number_u64 > self.canonicalization_delay {
-			let new_canonical = number_u64 - self.canonicalization_delay;
+		let best_canonical = || self.storage.state_db.best_canonical().unwrap_or(0);
+		let info = self.blockchain.info();
+		let best_number: u64 = self.blockchain.info().best_number.saturated_into();
 
-			if new_canonical <= self.storage.state_db.best_canonical().unwrap_or(0) {
-				return Ok(())
-			}
-			let hash = if new_canonical == number_u64 {
-				hash
-			} else {
-				sc_client_api::blockchain::HeaderBackend::hash(
-					&self.blockchain,
-					new_canonical.saturated_into(),
-				)?
-				.ok_or_else(|| {
-					sp_blockchain::Error::Backend(format!(
-						"Can't canonicalize missing block number #{} when importing {:?} (#{})",
-						new_canonical, hash, number,
-					))
-				})?
-			};
-			if !sc_client_api::Backend::have_state_at(self, hash, new_canonical.saturated_into()) {
+		while best_number.saturating_sub(best_canonical()) > self.canonicalization_delay {
+			let to_canonicalize = best_canonical() + 1;
+
+			let hash_to_canonicalize = sc_client_api::blockchain::HeaderBackend::hash(
+				&self.blockchain,
+				to_canonicalize.saturated_into(),
+			)?
+			.ok_or_else(|| {
+					let best_hash = info.best_hash;
+
+				sp_blockchain::Error::Backend(format!(
+					"Can't canonicalize missing block number #{to_canonicalize} when for best block {best_hash:?} (#{best_number})",
+				))
+			})?;
+
+			if !sc_client_api::Backend::have_state_at(
+				self,
+				hash_to_canonicalize,
+				to_canonicalize.saturated_into(),
+			) {
 				return Ok(())
 			}
 
-			trace!(target: "db", "Canonicalize block #{} ({:?})", new_canonical, hash);
-			let commit = self.storage.state_db.canonicalize_block(&hash).map_err(
+			trace!(target: "db", "Canonicalize block #{} ({:?})", to_canonicalize, hash_to_canonicalize);
+			let commit = self.storage.state_db.canonicalize_block(&hash_to_canonicalize).map_err(
 				sp_blockchain::Error::from_state_db::<
 					sc_state_db::Error<sp_database::error::DatabaseError>,
 				>,
 			)?;
 			apply_state_commit(transaction, commit);
 		}
+
 		Ok(())
 	}
 
@@ -1559,7 +1559,7 @@ impl<Block: BlockT> Backend<Block> {
 				)?;
 			} else {
 				// canonicalize blocks which are old enough, regardless of finality.
-				self.force_delayed_canonicalize(&mut transaction, hash, *header.number())?
+				self.force_delayed_canonicalize(&mut transaction)?
 			}
 
 			if !existing_header {
@@ -2759,6 +2759,33 @@ pub(crate) mod tests {
 				.into();
 			let hash = header.hash();
 
+			op.set_block_data(header, Some(vec![]), None, None, NewBlockState::Best)
+				.unwrap();
+
+			backend.commit_operation(op).unwrap();
+			hash
+		};
+
+		let hashof4 = {
+			let mut op = backend.begin_operation().unwrap();
+			backend.begin_state_operation(&mut op, hashof3).unwrap();
+			let mut header = Header {
+				number: 4,
+				parent_hash: hashof3,
+				state_root: Default::default(),
+				digest: Default::default(),
+				extrinsics_root: Default::default(),
+			};
+
+			let storage: Vec<(_, _)> = vec![];
+
+			header.state_root = op
+				.old_state
+				.storage_root(storage.iter().cloned().map(|(x, y)| (x, Some(y))), state_version)
+				.0
+				.into();
+			let hash = header.hash();
+
 			op.set_block_data(header, Some(vec![]), None, None, NewBlockState::Best)
 				.unwrap();
 
@@ -2774,6 +2801,7 @@ pub(crate) mod tests {
 		backend.finalize_block(hashof1, None).unwrap();
 		backend.finalize_block(hashof2, None).unwrap();
 		backend.finalize_block(hashof3, None).unwrap();
+		backend.finalize_block(hashof4, None).unwrap();
 		assert!(backend
 			.storage
 			.db
@@ -3801,4 +3829,105 @@ pub(crate) mod tests {
 		assert_eq!(backend.blockchain().leaves().unwrap(), vec![block2]);
 		assert_eq!(backend.blockchain().info().best_hash, block2);
 	}
+
+	#[test]
+	fn force_delayed_canonicalize_waiting_for_blocks_to_be_finalized() {
+		let backend = Backend::<Block>::new_test(10, 1);
+
+		let genesis =
+			insert_block(&backend, 0, Default::default(), None, Default::default(), vec![], None)
+				.unwrap();
+
+		let block1 = {
+			let mut op = backend.begin_operation().unwrap();
+			backend.begin_state_operation(&mut op, genesis).unwrap();
+			let header = Header {
+				number: 1,
+				parent_hash: genesis,
+				state_root: BlakeTwo256::trie_root(Vec::new(), StateVersion::V1),
+				digest: Default::default(),
+				extrinsics_root: Default::default(),
+			};
+
+			op.set_block_data(header.clone(), Some(Vec::new()), None, None, NewBlockState::Normal)
+				.unwrap();
+
+			backend.commit_operation(op).unwrap();
+
+			header.hash()
+		};
+
+		assert_eq!(0, backend.storage.state_db.best_canonical().unwrap());
+
+		// This should not trigger any forced canonicalization as we didn't have imported any best
+		// block by now.
+		let block2 = {
+			let mut op = backend.begin_operation().unwrap();
+			backend.begin_state_operation(&mut op, block1).unwrap();
+			let header = Header {
+				number: 2,
+				parent_hash: block1,
+				state_root: BlakeTwo256::trie_root(Vec::new(), StateVersion::V1),
+				digest: Default::default(),
+				extrinsics_root: Default::default(),
+			};
+
+			op.set_block_data(header.clone(), Some(Vec::new()), None, None, NewBlockState::Normal)
+				.unwrap();
+
+			backend.commit_operation(op).unwrap();
+
+			header.hash()
+		};
+
+		assert_eq!(0, backend.storage.state_db.best_canonical().unwrap());
+
+		// This should also not trigger it yet, because we import a best block, but the best block
+		// from the POV of the db is still at `0`.
+		let block3 = {
+			let mut op = backend.begin_operation().unwrap();
+			backend.begin_state_operation(&mut op, block2).unwrap();
+			let header = Header {
+				number: 3,
+				parent_hash: block2,
+				state_root: BlakeTwo256::trie_root(Vec::new(), StateVersion::V1),
+				digest: Default::default(),
+				extrinsics_root: Default::default(),
+			};
+
+			op.set_block_data(header.clone(), Some(Vec::new()), None, None, NewBlockState::Best)
+				.unwrap();
+
+			backend.commit_operation(op).unwrap();
+
+			header.hash()
+		};
+
+		// Now it should kick in.
+		let block4 = {
+			let mut op = backend.begin_operation().unwrap();
+			backend.begin_state_operation(&mut op, block3).unwrap();
+			let header = Header {
+				number: 4,
+				parent_hash: block3,
+				state_root: BlakeTwo256::trie_root(Vec::new(), StateVersion::V1),
+				digest: Default::default(),
+				extrinsics_root: Default::default(),
+			};
+
+			op.set_block_data(header.clone(), Some(Vec::new()), None, None, NewBlockState::Best)
+				.unwrap();
+
+			backend.commit_operation(op).unwrap();
+
+			header.hash()
+		};
+
+		assert_eq!(2, backend.storage.state_db.best_canonical().unwrap());
+
+		assert_eq!(block1, backend.blockchain().hash(1).unwrap().unwrap());
+		assert_eq!(block2, backend.blockchain().hash(2).unwrap().unwrap());
+		assert_eq!(block3, backend.blockchain().hash(3).unwrap().unwrap());
+		assert_eq!(block4, backend.blockchain().hash(4).unwrap().unwrap());
+	}
 }