add warp_sync_params

.gitlab-ci.yml

@@ -30,7 +30,7 @@ variables:
   CI_IMAGE:                        "paritytech/ci-linux:production"
   DOCKER_OS:                       "debian:stretch"
   ARCH:                            "x86_64"
-  ZOMBIENET_IMAGE:                 "docker.io/paritytech/zombienet:v1.3.29"
+  ZOMBIENET_IMAGE:                 "docker.io/paritytech/zombienet:v1.3.34"
 
 .collect-artifacts:
   artifacts:

client/consensus/common/src/tests.rs

@@ -85,6 +85,10 @@ impl RelayChainInterface for Relaychain {
 		unimplemented!("Not needed for test")
 	}
 
+	async fn finalized_block_hash(&self) -> RelayChainResult<PHash> {
+		unimplemented!("Not needed for test")
+	}
+
 	async fn retrieve_dmq_contents(
 		&self,
 		_: ParaId,

client/network/src/lib.rs

@@ -36,7 +36,6 @@ use polkadot_primitives::{
 
 use codec::{Decode, DecodeAll, Encode};
 use futures::{channel::oneshot, future::FutureExt, Future};
-
 use std::{convert::TryFrom, fmt, marker::PhantomData, pin::Pin, sync::Arc};
 
 #[cfg(test)]

client/network/src/tests.rs

@@ -84,6 +84,9 @@ impl RelayChainInterface for DummyRelayChainInterface {
 	async fn best_block_hash(&self) -> RelayChainResult<PHash> {
 		Ok(self.relay_backend.blockchain().info().best_hash)
 	}
+	async fn finalized_block_hash(&self) -> RelayChainResult<PHash> {
+		Ok(self.relay_backend.blockchain().info().finalized_hash)
+	}
 
 	async fn retrieve_dmq_contents(
 		&self,

client/relay-chain-inprocess-interface/src/lib.rs

@@ -167,6 +167,10 @@ where
 		Ok(self.backend.blockchain().info().best_hash)
 	}
 
+	async fn finalized_block_hash(&self) -> RelayChainResult<PHash> {
+		Ok(self.backend.blockchain().info().finalized_hash)
+	}
+
 	async fn is_major_syncing(&self) -> RelayChainResult<bool> {
 		Ok(self.sync_oracle.is_major_syncing())
 	}

client/relay-chain-interface/src/lib.rs

@@ -105,6 +105,9 @@ pub trait RelayChainInterface: Send + Sync {
 	/// Get the hash of the current best block.
 	async fn best_block_hash(&self) -> RelayChainResult<PHash>;
 
+	/// Get the hash of the finalized block.
+	async fn finalized_block_hash(&self) -> RelayChainResult<PHash>;
+
 	/// Returns the whole contents of the downward message queue for the parachain we are collating
 	/// for.
 	///
@@ -248,6 +251,10 @@ where
 		(**self).best_block_hash().await
 	}
 
+	async fn finalized_block_hash(&self) -> RelayChainResult<PHash> {
+		(**self).finalized_block_hash().await
+	}
+
 	async fn is_major_syncing(&self) -> RelayChainResult<bool> {
 		(**self).is_major_syncing().await
 	}

client/relay-chain-rpc-interface/src/lib.rs

@@ -123,6 +123,10 @@ impl RelayChainInterface for RelayChainRpcInterface {
 		self.rpc_client.chain_get_head(None).await
 	}
 
+	async fn finalized_block_hash(&self) -> RelayChainResult<RelayHash> {
+		self.rpc_client.chain_get_finalized_head().await
+	}
+
 	async fn is_major_syncing(&self) -> RelayChainResult<bool> {
 		self.rpc_client.system_health().await.map(|h| h.is_syncing)
 	}

client/service/Cargo.toml

@@ -11,14 +11,20 @@ futures = "0.3.26"
 # Substrate
 sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-transaction-pool = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-rpc = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-service = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-sysinfo = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-telemetry = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-utils = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-network-transactions = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-transaction-pool = { git = "https://github.com/paritytech/substrate", branch = "master" }
 
 # Polkadot
 polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "master" }
@@ -28,7 +34,8 @@ cumulus-client-cli = { path = "../cli" }
 cumulus-client-collator = { path = "../collator" }
 cumulus-client-consensus-common = { path = "../consensus/common" }
 cumulus-client-pov-recovery = { path = "../pov-recovery" }
+cumulus-client-network = { path = "../network" }
 cumulus-primitives-core = { path = "../../primitives/core" }
 cumulus-relay-chain-interface = { path = "../relay-chain-interface" }
 cumulus-relay-chain-inprocess-interface = { path = "../relay-chain-inprocess-interface" }
-cumulus-relay-chain-minimal-node = { path = "../relay-chain-minimal-node" }
+cumulus-relay-chain-minimal-node = { path = "../relay-chain-minimal-node" }

client/service/src/lib.rs

@@ -20,31 +20,37 @@
 
 use cumulus_client_cli::CollatorOptions;
 use cumulus_client_consensus_common::ParachainConsensus;
+use cumulus_client_network::BlockAnnounceValidator;
 use cumulus_client_pov_recovery::{PoVRecovery, RecoveryDelayRange, RecoveryHandle};
 use cumulus_primitives_core::{CollectCollationInfo, ParaId};
 use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain;
 use cumulus_relay_chain_interface::{RelayChainInterface, RelayChainResult};
 use cumulus_relay_chain_minimal_node::build_minimal_relay_chain_node;
-use polkadot_primitives::CollatorPair;
-
+use futures::{
+	channel::{mpsc, oneshot},
+	FutureExt, StreamExt,
+};
+use polkadot_primitives::{CollatorPair, OccupiedCoreAssumption};
 use sc_client_api::{
-	Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, UsageProvider,
+	Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, ProofProvider, UsageProvider,
 };
-use sc_consensus::{import_queue::ImportQueueService, BlockImport};
-use sc_service::{Configuration, TaskManager};
-use sc_telemetry::TelemetryWorkerHandle;
+use sc_consensus::{import_queue::ImportQueueService, BlockImport, ImportQueue};
+use sc_network::{config::SyncMode, NetworkService};
+use sc_network_transactions::TransactionsHandlerController;
+use sc_service::{Configuration, NetworkStarter, SpawnTaskHandle, TaskManager, WarpSyncParams};
+use sc_telemetry::{log, TelemetryWorkerHandle};
+use sc_utils::mpsc::TracingUnboundedSender;
 use sp_api::ProvideRuntimeApi;
-use sp_blockchain::HeaderBackend;
-use sp_core::traits::SpawnNamed;
-use sp_runtime::traits::Block as BlockT;
-
-use futures::channel::mpsc;
+use sp_blockchain::{HeaderBackend, HeaderMetadata};
+use sp_core::{traits::SpawnNamed, Decode};
+use sp_runtime::traits::{Block as BlockT, BlockIdTo};
 use std::{sync::Arc, time::Duration};
 
 // Given the sporadic nature of the explicit recovery operation and the
 // possibility to retry infinite times this value is more than enough.
 // In practice here we expect no more than one queued messages.
 const RECOVERY_CHAN_SIZE: usize = 8;
+const LOG_TARGET_SYNC: &str = "sync::cumulus";
 
 /// Parameters given to [`start_collator`].
 pub struct StartCollatorParams<'a, Block: BlockT, BS, Client, RCInterface, Spawner> {
@@ -268,3 +274,172 @@ pub async fn build_relay_chain_interface(
 		)
 	}
 }
+
+/// Parameters given to [`build_network`].
+pub struct BuildNetworkParams<
+	'a,
+	Block: BlockT,
+	Client: ProvideRuntimeApi<Block>
+		+ BlockBackend<Block>
+		+ HeaderMetadata<Block, Error = sp_blockchain::Error>
+		+ HeaderBackend<Block>
+		+ BlockIdTo<Block>
+		+ 'static,
+	RCInterface,
+	IQ,
+> where
+	Client::Api: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>,
+{
+	pub parachain_config: &'a Configuration,
+	pub client: Arc<Client>,
+	pub transaction_pool: Arc<sc_transaction_pool::FullPool<Block, Client>>,
+	pub para_id: ParaId,
+	pub relay_chain_interface: RCInterface,
+	pub spawn_handle: SpawnTaskHandle,
+	pub import_queue: IQ,
+}
+
+/// Build the network service, the network status sinks and an RPC sender.
+pub async fn build_network<'a, Block, Client, RCInterface, IQ>(
+	BuildNetworkParams {
+		parachain_config,
+		client,
+		transaction_pool,
+		para_id,
+		spawn_handle,
+		relay_chain_interface,
+		import_queue,
+	}: BuildNetworkParams<'a, Block, Client, RCInterface, IQ>,
+) -> sc_service::error::Result<(
+	Arc<NetworkService<Block, Block::Hash>>,
+	TracingUnboundedSender<sc_rpc::system::Request<Block>>,
+	TransactionsHandlerController<Block::Hash>,
+	NetworkStarter,
+)>
+where
+	Block: BlockT,
+	Client: UsageProvider<Block>
+		+ HeaderBackend<Block>
+		+ sp_consensus::block_validation::Chain<Block>
+		+ Send
+		+ Sync
+		+ BlockBackend<Block>
+		+ BlockchainEvents<Block>
+		+ ProvideRuntimeApi<Block>
+		+ HeaderMetadata<Block, Error = sp_blockchain::Error>
+		+ BlockIdTo<Block, Error = sp_blockchain::Error>
+		+ ProofProvider<Block>
+		+ 'static,
+	Client::Api: CollectCollationInfo<Block>
+		+ sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>,
+	for<'b> &'b Client: BlockImport<Block>,
+	RCInterface: RelayChainInterface + Clone + 'static,
+	IQ: ImportQueue<Block> + 'static,
+{
+	let warp_sync_params = match parachain_config.network.sync_mode {
+		SyncMode::Warp => {
+			let target_block = warp_sync_get::<Block, RCInterface>(
+				para_id,
+				relay_chain_interface.clone(),
+				spawn_handle.clone(),
+			);
+			Some(WarpSyncParams::WaitForTarget(target_block))
+		},
+		_ => None,
+	};
+
+	let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface, para_id);
+	let block_announce_validator_builder = move |_| Box::new(block_announce_validator) as Box<_>;
+
+	sc_service::build_network(sc_service::BuildNetworkParams {
+		config: parachain_config,
+		client,
+		transaction_pool,
+		spawn_handle,
+		import_queue,
+		block_announce_validator_builder: Some(Box::new(block_announce_validator_builder)),
+		warp_sync_params,
+	})
+}
+
+/// Creates a new background task to wait for the relay chain to sync up and retrieve the parachain header
+fn warp_sync_get<B, RCInterface>(
+	para_id: ParaId,
+	relay_chain_interface: RCInterface,
+	spawner: SpawnTaskHandle,
+) -> oneshot::Receiver<<B as BlockT>::Header>
+where
+	B: BlockT + 'static,
+	RCInterface: RelayChainInterface + 'static,
+{
+	let (sender, receiver) = oneshot::channel::<B::Header>();
+	spawner.spawn(
+		"cumulus-parachain-wait-for-target-block",
+		None,
+		async move {
+			log::debug!(
+				target: "cumulus-network",
+				"waiting for announce block in a background task...",
+			);
+
+			let _ = wait_for_target_block::<B, _>(sender, para_id, relay_chain_interface)
+				.await
+				.map_err(|e| {
+					log::error!(
+						target: LOG_TARGET_SYNC,
+						"Unable to determine parachain target block {:?}",
+						e
+					)
+				});
+		}
+		.boxed(),
+	);
+
+	receiver
+}
+
+/// Waits for the relay chain to have finished syncing and then gets the parachain header that corresponds to the last finalized relay chain block.
+async fn wait_for_target_block<B, RCInterface>(
+	sender: oneshot::Sender<<B as BlockT>::Header>,
+	para_id: ParaId,
+	relay_chain_interface: RCInterface,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
+where
+	B: BlockT + 'static,
+	RCInterface: RelayChainInterface + Send + 'static,
+{
+	let mut imported_blocks = relay_chain_interface.import_notification_stream().await?.fuse();
+	while imported_blocks.next().await.is_some() {
+		let is_syncing = relay_chain_interface.is_major_syncing().await.map_err(|e| {
+			Box::<dyn std::error::Error + Send + Sync>::from(format!(
+				"Unable to determine sync status. {e}"
+			))
+		})?;
+
+		if !is_syncing {
+			let relay_chain_best_hash = relay_chain_interface
+				.finalized_block_hash()
+				.await
+				.map_err(|e| Box::new(e) as Box<_>)?;
+
+			let validation_data = relay_chain_interface
+				.persisted_validation_data(
+					relay_chain_best_hash,
+					para_id,
+					OccupiedCoreAssumption::TimedOut,
+				)
+				.await
+				.map_err(|e| format!("{e:?}"))?
+				.ok_or_else(|| "Could not find parachain head in relay chain")?;
+
+			let target_block = B::Header::decode(&mut &validation_data.parent_head.0[..])
+				.map_err(|e| format!("Failed to decode parachain head: {e}"))?;
+
+			log::debug!(target: LOG_TARGET_SYNC, "Target block reached {:?}", target_block);
+			let _ = sender.send(target_block);
+			return Ok(())
+		}
+	}
+
+	Err("Stopping following imported blocks. Could not determine parachain target block".into())
+}

parachain-template/node/src/service.rs

@@ -12,10 +12,9 @@ use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, Slo
 use cumulus_client_consensus_common::{
 	ParachainBlockImport as TParachainBlockImport, ParachainConsensus,
 };
-use cumulus_client_network::BlockAnnounceValidator;
 use cumulus_client_service::{
-	build_relay_chain_interface, prepare_node_config, start_collator, start_full_node,
-	StartCollatorParams, StartFullNodeParams,
+	build_network, build_relay_chain_interface, prepare_node_config, start_collator,
+	start_full_node, BuildNetworkParams, StartCollatorParams, StartFullNodeParams,
 };
 use cumulus_primitives_core::ParaId;
 use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface};
@@ -168,27 +167,23 @@ async fn start_node_impl(
 		s => s.to_string().into(),
 	})?;
 
-	let block_announce_validator =
-		BlockAnnounceValidator::new(relay_chain_interface.clone(), para_id);
-
 	let force_authoring = parachain_config.force_authoring;
 	let validator = parachain_config.role.is_authority();
 	let prometheus_registry = parachain_config.prometheus_registry().cloned();
 	let transaction_pool = params.transaction_pool.clone();
 	let import_queue_service = params.import_queue.service();
 
 	let (network, system_rpc_tx, tx_handler_controller, start_network) =
-		sc_service::build_network(sc_service::BuildNetworkParams {
-			config: &parachain_config,
+		build_network(BuildNetworkParams {
+			parachain_config: &parachain_config,
 			client: client.clone(),
 			transaction_pool: transaction_pool.clone(),
+			para_id,
 			spawn_handle: task_manager.spawn_handle(),
+			relay_chain_interface: relay_chain_interface.clone(),
 			import_queue: params.import_queue,
-			block_announce_validator_builder: Some(Box::new(|_| {
-				Box::new(block_announce_validator)
-			})),
-			warp_sync: None,
-		})?;
+		})
+		.await?;
 
 	if parachain_config.offchain_worker.enabled {
 		sc_service::build_offchain_workers(