add pov-recovery unit tests and support for elastic scaling (#4733)

- unit tests for pov-recovery
- elastic scaling support (recovering multiple candidates in a single
relay chain block)
- also some small cleanups
- also switches to candidates_pending_availability in
`handle_empty_block_announce_data`

Fixes #3577

After #4097 is merged, we
should also add a zombienet test, similar to the existing
`0002-pov_recovery.toml` but which has a single collator using elastic
scaling on multiple cores.

cumulus/client/consensus/common/Cargo.toml

@@ -28,6 +28,7 @@ sp-core = { path = "../../../../substrate/primitives/core" }
 sp-runtime = { path = "../../../../substrate/primitives/runtime" }
 sp-timestamp = { path = "../../../../substrate/primitives/timestamp" }
 sp-trie = { path = "../../../../substrate/primitives/trie" }
+sp-version = { path = "../../../../substrate/primitives/version" }
 prometheus-endpoint = { package = "substrate-prometheus-endpoint", path = "../../../../substrate/utils/prometheus" }
 
 # Polkadot

cumulus/client/consensus/common/src/tests.rs

@@ -38,6 +38,7 @@ use polkadot_primitives::HeadData;
 use sc_client_api::{Backend as _, UsageProvider};
 use sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
 use sp_consensus::{BlockOrigin, BlockStatus};
+use sp_version::RuntimeVersion;
 use std::{
 	collections::{BTreeMap, HashMap},
 	pin::Pin,
@@ -153,6 +154,14 @@ impl RelayChainInterface for Relaychain {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn candidates_pending_availability(
+		&self,
+		_: PHash,
+		_: ParaId,
+	) -> RelayChainResult<Vec<CommittedCandidateReceipt>> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn session_index_for_child(&self, _: PHash) -> RelayChainResult<SessionIndex> {
 		Ok(0)
 	}
@@ -247,6 +256,10 @@ impl RelayChainInterface for Relaychain {
 			extrinsics_root: PHash::zero(),
 		}))
 	}
+
+	async fn version(&self, _: PHash) -> RelayChainResult<RuntimeVersion> {
+		unimplemented!("Not needed for test")
+	}
 }
 
 fn sproof_with_best_parent(client: &Client) -> RelayStateSproofBuilder {

cumulus/client/network/Cargo.toml

@@ -24,11 +24,14 @@ sp-consensus = { path = "../../../substrate/primitives/consensus/common" }
 sp-core = { path = "../../../substrate/primitives/core" }
 sp-runtime = { path = "../../../substrate/primitives/runtime" }
 sp-state-machine = { path = "../../../substrate/primitives/state-machine" }
+sp-api = { path = "../../../substrate/primitives/api" }
+sp-version = { path = "../../../substrate/primitives/version" }
 
 # Polkadot
 polkadot-node-primitives = { path = "../../../polkadot/node/primitives" }
 polkadot-parachain-primitives = { path = "../../../polkadot/parachain" }
 polkadot-primitives = { path = "../../../polkadot/primitives" }
+polkadot-node-subsystem = { path = "../../../polkadot/node/subsystem" }
 
 # Cumulus
 cumulus-relay-chain-interface = { path = "../relay-chain-interface" }
@@ -37,6 +40,7 @@ cumulus-relay-chain-interface = { path = "../relay-chain-interface" }
 portpicker = "0.1.1"
 tokio = { version = "1.32.0", features = ["macros"] }
 url = "2.4.0"
+rstest = "0.18.2"
 
 # Substrate
 sc-cli = { path = "../../../substrate/client/cli" }

cumulus/client/network/src/lib.rs

@@ -20,6 +20,7 @@
 //! that use the relay chain provided consensus. See [`RequireSecondedInBlockAnnounce`]
 //! and [`WaitToAnnounce`] for more information about this implementation.
 
+use sp_api::RuntimeApiInfo;
 use sp_consensus::block_validation::{
 	BlockAnnounceValidator as BlockAnnounceValidatorT, Validation,
 };
@@ -28,6 +29,7 @@ use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 
 use cumulus_relay_chain_interface::RelayChainInterface;
 use polkadot_node_primitives::{CollationSecondedSignal, Statement};
+use polkadot_node_subsystem::messages::RuntimeApiRequest;
 use polkadot_parachain_primitives::primitives::HeadData;
 use polkadot_primitives::{
 	CandidateReceipt, CompactStatement, Hash as PHash, Id as ParaId, OccupiedCoreAssumption,
@@ -266,18 +268,41 @@ where
 		Ok(para_head)
 	}
 
-	/// Get the backed block hash of the given parachain in the relay chain.
-	async fn backed_block_hash(
+	/// Get the backed block hashes of the given parachain in the relay chain.
+	async fn backed_block_hashes(
 		relay_chain_interface: &RCInterface,
 		hash: PHash,
 		para_id: ParaId,
-	) -> Result<Option<PHash>, BoxedError> {
-		let candidate_receipt = relay_chain_interface
-			.candidate_pending_availability(hash, para_id)
+	) -> Result<impl Iterator<Item = PHash>, BoxedError> {
+		let runtime_api_version = relay_chain_interface
+			.version(hash)
 			.await
 			.map_err(|e| Box::new(BlockAnnounceError(format!("{:?}", e))) as Box<_>)?;
+		let parachain_host_runtime_api_version =
+			runtime_api_version
+				.api_version(
+					&<dyn polkadot_primitives::runtime_api::ParachainHost<
+						polkadot_primitives::Block,
+					>>::ID,
+				)
+				.unwrap_or_default();
+
+		// If the relay chain runtime does not support the new runtime API, fallback to the
+		// deprecated one.
+		let candidate_receipts = if parachain_host_runtime_api_version <
+			RuntimeApiRequest::CANDIDATES_PENDING_AVAILABILITY_RUNTIME_REQUIREMENT
+		{
+			#[allow(deprecated)]
+			relay_chain_interface
+				.candidate_pending_availability(hash, para_id)
+				.await
+				.map(|c| c.into_iter().collect::<Vec<_>>())
+		} else {
+			relay_chain_interface.candidates_pending_availability(hash, para_id).await
+		}
+		.map_err(|e| Box::new(BlockAnnounceError(format!("{:?}", e))) as Box<_>)?;
 
-		Ok(candidate_receipt.map(|cr| cr.descriptor.para_head))
+		Ok(candidate_receipts.into_iter().map(|cr| cr.descriptor.para_head))
 	}
 
 	/// Handle a block announcement with empty data (no statement) attached to it.
@@ -298,15 +323,20 @@ where
 		let best_head =
 			Self::included_block(&relay_chain_interface, relay_chain_best_hash, para_id).await?;
 		let known_best_number = best_head.number();
-		let backed_block = || async {
-			Self::backed_block_hash(&relay_chain_interface, relay_chain_best_hash, para_id).await
-		};
 
 		if best_head == header {
 			tracing::debug!(target: LOG_TARGET, "Announced block matches best block.",);
 
-			Ok(Validation::Success { is_new_best: true })
-		} else if Some(HeadData(header.encode()).hash()) == backed_block().await? {
+			return Ok(Validation::Success { is_new_best: true })
+		}
+
+		let mut backed_blocks =
+			Self::backed_block_hashes(&relay_chain_interface, relay_chain_best_hash, para_id)
+				.await?;
+
+		let head_hash = HeadData(header.encode()).hash();
+
+		if backed_blocks.any(|block_hash| block_hash == head_hash) {
 			tracing::debug!(target: LOG_TARGET, "Announced block matches latest backed block.",);
 
 			Ok(Validation::Success { is_new_best: true })