wait for relay chain to sync then get parachain header

Author: samelamin

Cargo.lock

@@ -18,7 +18,7 @@ use polkadot_primitives::v2::{Hash as PHash, PersistedValidationData};
 use sc_consensus::BlockImport;
 use sp_runtime::traits::Block as BlockT;
 
-mod parachain_consensus;
+pub mod parachain_consensus;
 #[cfg(test)]
 mod tests;
 pub use parachain_consensus::run_parachain_consensus;

client/consensus/common/src/lib.rs

@@ -28,6 +28,7 @@ polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch =
 
 # Cumulus
 cumulus-relay-chain-interface = { path = "../relay-chain-interface" }
+cumulus-client-consensus-common = { path = "../consensus/common" }
 
 [dev-dependencies]
 portpicker = "0.1.1"

client/network/Cargo.toml

@@ -35,10 +35,10 @@ use polkadot_primitives::v2::{
 };
 
 use codec::{Decode, DecodeAll, Encode};
-use futures::{channel::oneshot, future::FutureExt, Future};
+use futures::{channel::oneshot, future::FutureExt, Future, StreamExt};
 
+use cumulus_client_consensus_common::parachain_consensus::RelaychainClient;
 use std::{convert::TryFrom, fmt, marker::PhantomData, pin::Pin, sync::Arc};
-
 #[cfg(test)]
 mod tests;
 
@@ -454,3 +454,72 @@ async fn wait_to_announce<Block: BlockT>(
 		);
 	}
 }
+
+#[derive(Clone)]
+pub struct WaitForParachainTargetBlock<Block> {
+	phantom: PhantomData<Block>,
+}
+
+impl<Block: BlockT> WaitForParachainTargetBlock<Block> {
+	/// Get warp sync target block
+	pub async fn warp_sync_get(
+		para_id: ParaId,
+		relay_chain_interface: Arc<dyn RelayChainInterface>,
+	) -> Result<oneshot::Receiver<Block::Header>, BoxedError>
+	where
+		Block: BlockT + 'static,
+	{
+		let (sender, receiver) = oneshot::channel::<Block::Header>();
+		Self::wait_for_target_block(sender, para_id, relay_chain_interface).await;
+		return Ok(receiver)
+	}
+
+	async fn wait_for_target_block(
+		sender: oneshot::Sender<Block::Header>,
+		para_id: ParaId,
+		relay_chain_interface: Arc<dyn RelayChainInterface>,
+	) {
+		let is_syncing = relay_chain_interface
+			.is_major_syncing()
+			.await
+			.map_err(|e| {
+				tracing::error!(target: LOG_TARGET, "Unable to determine sync status. {}", e)
+			})
+			.unwrap_or(false);
+
+		loop {
+			if !is_syncing {
+				let mut finalized_heads = match relay_chain_interface.finalized_heads(para_id).await
+				{
+					Ok(finalized_heads_stream) => finalized_heads_stream,
+					Err(err) => {
+						tracing::error!(target: LOG_TARGET, error = ?err, "Unable to retrieve finalized heads stream.");
+						return
+					},
+				};
+
+				let finalized_head = if let Some(h) = finalized_heads.next().await {
+					h
+				} else {
+					tracing::debug!(target: "cumulus-consensus", "Stopping following finalized head.");
+					return
+				};
+
+				let target_header = match Block::Header::decode(&mut &finalized_head[..]) {
+					Ok(header) => header,
+					Err(err) => {
+						tracing::debug!(
+							target: "cumulus-network",
+							error = ?err,
+							"Could not decode parachain header while following finalized heads.",
+						);
+						continue
+					},
+				};
+
+				let _ = sender.send(target_header);
+				break
+			}
+		}
+	}
+}

client/network/src/lib.rs

@@ -31,6 +31,7 @@ use sp_keystore::SyncCryptoStorePtr;
 use substrate_prometheus_endpoint::Registry;
 
 use polkadot_service::CollatorPair;
+use sc_network_common::sync::warp::WarpSyncParams;
 
 /// Native executor type.
 pub struct ParachainNativeExecutor;
@@ -197,6 +198,17 @@ async fn start_node_impl(
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
 	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);
+	let warp_sync_params =
+		match cumulus_client_network::WaitForParachainTargetBlock::<Block>::warp_sync_get(
+			id,
+			relay_chain_interface.clone(),
+		)
+		.await
+		{
+			Ok(target_block) => Some(WarpSyncParams::WaitForTarget(target_block)),
+			_ => None,
+		};
+
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &parachain_config,
@@ -207,7 +219,7 @@ async fn start_node_impl(
 			block_announce_validator_builder: Some(Box::new(|_| {
 				Box::new(block_announce_validator)
 			})),
-			warp_sync_params: None,
+			warp_sync_params,
 		})?;
 
 	if parachain_config.offchain_worker.enabled {

parachain-template/node/src/service.rs

@@ -47,7 +47,7 @@ use sc_consensus::{
 };
 use sc_executor::WasmExecutor;
 use sc_network::NetworkService;
-use sc_network_common::service::NetworkBlock;
+use sc_network_common::{service::NetworkBlock, sync::warp::WarpSyncParams};
 use sc_service::{Configuration, PartialComponents, TFullBackend, TFullClient, TaskManager};
 use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};
 use sp_api::{ApiExt, ConstructRuntimeApi};
@@ -375,6 +375,16 @@ where
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
 	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);
+	let warp_sync_params =
+		match cumulus_client_network::WaitForParachainTargetBlock::<Block>::warp_sync_get(
+			para_id,
+			relay_chain_interface.clone(),
+		)
+		.await
+		{
+			Ok(target_block) => Some(WarpSyncParams::WaitForTarget(target_block)),
+			_ => None,
+		};
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &parachain_config,
@@ -385,7 +395,7 @@ where
 			block_announce_validator_builder: Some(Box::new(|_| {
 				Box::new(block_announce_validator)
 			})),
-			warp_sync_params: None,
+			warp_sync_params,
 		})?;
 
 	let rpc_client = client.clone();
@@ -558,6 +568,16 @@ where
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
 	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);
+	let warp_sync_params =
+		match cumulus_client_network::WaitForParachainTargetBlock::<Block>::warp_sync_get(
+			para_id,
+			relay_chain_interface.clone(),
+		)
+		.await
+		{
+			Ok(target_block) => Some(WarpSyncParams::WaitForTarget(target_block)),
+			_ => None,
+		};
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &parachain_config,
@@ -568,7 +588,7 @@ where
 			block_announce_validator_builder: Some(Box::new(|_| {
 				Box::new(block_announce_validator)
 			})),
-			warp_sync_params: None,
+			warp_sync_params,
 		})?;
 
 	let rpc_builder = {
@@ -1327,6 +1347,16 @@ where
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
 	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);
+	let warp_sync_params =
+		match cumulus_client_network::WaitForParachainTargetBlock::<Block>::warp_sync_get(
+			para_id,
+			relay_chain_interface.clone(),
+		)
+		.await
+		{
+			Ok(target_block) => Some(WarpSyncParams::WaitForTarget(target_block)),
+			_ => None,
+		};
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &parachain_config,
@@ -1337,7 +1367,7 @@ where
 			block_announce_validator_builder: Some(Box::new(|_| {
 				Box::new(block_announce_validator)
 			})),
-			warp_sync_params: None,
+			warp_sync_params,
 		})?;
 
 	let rpc_builder = {

polkadot-parachain/src/service.rs

@@ -49,7 +49,9 @@ use polkadot_primitives::v2::{CollatorPair, Hash as PHash, PersistedValidationDa
 use polkadot_service::ProvideRuntimeApi;
 use sc_client_api::execution_extensions::ExecutionStrategies;
 use sc_network::{multiaddr, NetworkBlock, NetworkService};
-use sc_network_common::{config::TransportConfig, service::NetworkStateInfo};
+use sc_network_common::{
+	config::TransportConfig, service::NetworkStateInfo, sync::warp::WarpSyncParams,
+};
 use sc_service::{
 	config::{
 		BlocksPruning, DatabaseSource, KeystoreConfig, MultiaddrWithPeerId, NetworkConfiguration,
@@ -268,6 +270,16 @@ where
 		BlockAnnounceValidator::new(relay_chain_interface.clone(), para_id);
 	let block_announce_validator_builder = move |_| Box::new(block_announce_validator) as Box<_>;
 
+	let warp_sync_params =
+		match cumulus_client_network::WaitForParachainTargetBlock::<Block>::warp_sync_get(
+			para_id,
+			relay_chain_interface.clone(),
+		)
+		.await
+		{
+			Ok(target_block) => Some(WarpSyncParams::WaitForTarget(target_block)),
+			_ => None,
+		};
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
@@ -278,7 +290,7 @@ where
 			spawn_handle: task_manager.spawn_handle(),
 			import_queue: import_queue.clone(),
 			block_announce_validator_builder: Some(Box::new(block_announce_validator_builder)),
-			warp_sync_params: None,
+			warp_sync_params,
 		})?;
 
 	let rpc_builder = {