add warp_sync_params (#1909)

* wait for relay chain to sync then get parachain header * Spawn new thread to wait for the target block * second round of comments from the PR on substrate * third round of pr comments * add zombienet tests * rebase issues * refactor tests based on pr comments * rebase issues * pr comments * passing zombienet test * cargo +nightly fmt * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * use cargo lock from master * pr comments * cargo fmt * use finalised block instead of best block * use import notification stream * rebase changes * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/relay-chain-interface/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/relay-chain-interface/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * pr comments * use new file names * db snaphots moved to google cloud storage * Update client/network/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Bastian Köcher <git@kchr.de> * Update client/service/src/lib.rs Co-authored-by: Sebastian Kunert <skunert49@gmail.com> * pr comments * Update zombienet/tests/0007-full_node_warp_sync.toml Co-authored-by: Sebastian Kunert <skunert49@gmail.com> * Update zombienet/tests/0007-full_node_warp_sync.toml Co-authored-by: Sebastian Kunert <skunert49@gmail.com> * Scenario 1 Parachain node and in-node relay chain both start with --sync warp. This ensures that the waiting logic works as expected. Scenario 2 Parachain node starts with warp sync, relay chain points to a node already synced up scenario 3 Parachain node starts with warp sync, relay chain points to a node that uses warp sync * Use test-parachain * use test-parachain chainspecs * remove relay chain spec as it is no longer required * add back relaychain spec file * pr comments * Upload snapshots to google cloud * Update zombienet/tests/0007-prepare-warp-sync-db-snapshot.md Co-authored-by: Sebastian Kunert <skunert49@gmail.com> * update documentation * Fix snapshot URLs * use master lock file * add finalized_block_hash * Patch diener for CI * Bump Zombienet * Add 0007 zombienet test * Bump zombienet * Revert "Patch diener for CI" This reverts commit 9ece6c9fc9b17058b61cd7e9dee29d3a9af87841. * merge fixes * use master lock file * Update Substrate & Polkadot --------- Co-authored-by: Bastian Köcher <git@kchr.de> Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Co-authored-by: Sebastian Kunert <skunert49@gmail.com> Co-authored-by: Bastian Köcher <info@kchr.de>
2026-06-13 01:11:10 +00:00 · 2023-02-14 19:15:43 +00:00
parent d6b16afb8c
commit f2dc402cec
19 changed files with 900 additions and 365 deletions
@@ -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 sc_client_api::{
-	Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, UsageProvider,
+use futures::{
+	channel::{mpsc, oneshot},
+	FutureExt, StreamExt,
 };
-use sc_consensus::{import_queue::ImportQueueService, BlockImport};
-use sc_service::{Configuration, TaskManager};
-use sc_telemetry::TelemetryWorkerHandle;
+use polkadot_primitives::{CollatorPair, OccupiedCoreAssumption};
+use sc_client_api::{
+	Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, ProofProvider, UsageProvider,
+};
+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())
+}