diff --git a/bin/node-template/node/src/service.rs b/bin/node-template/node/src/service.rs index ee8464688c79c..d7d075efd98ab 100644 --- a/bin/node-template/node/src/service.rs +++ b/bin/node-template/node/src/service.rs @@ -6,7 +6,7 @@ use sc_consensus_aura::{ImportQueueParams, SlotProportion, StartAuraParams}; pub use sc_executor::NativeElseWasmExecutor; use sc_finality_grandpa::SharedVoterState; use sc_keystore::LocalKeystore; -use sc_service::{error::Error as ServiceError, Configuration, TaskManager}; +use sc_service::{error::Error as ServiceError, Configuration, TaskManager, WarpSyncParams}; use sc_telemetry::{Telemetry, TelemetryWorker}; use sp_consensus_aura::sr25519::AuthorityPair as AuraPair; use std::{sync::Arc, time::Duration}; @@ -200,7 +200,7 @@ pub fn new_full(mut config: Configuration) -> Result spawn_handle: task_manager.spawn_handle(), import_queue, block_announce_validator_builder: None, - warp_sync: Some(warp_sync), + warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)), })?; if config.offchain_worker.enabled { diff --git a/bin/node/cli/src/service.rs b/bin/node/cli/src/service.rs index b1e9360b4a6c4..7ad1a9486b544 100644 --- a/bin/node/cli/src/service.rs +++ b/bin/node/cli/src/service.rs @@ -32,7 +32,9 @@ use sc_client_api::BlockBackend; use sc_consensus_babe::{self, SlotProportion}; use sc_executor::NativeElseWasmExecutor; use sc_network::NetworkService; -use sc_network_common::{protocol::event::Event, service::NetworkEventStream}; +use sc_network_common::{ + protocol::event::Event, service::NetworkEventStream, sync::warp::WarpSyncParams, +}; use sc_service::{config::Configuration, error::Error as ServiceError, RpcHandlers, TaskManager}; use sc_telemetry::{Telemetry, TelemetryWorker}; use sp_api::ProvideRuntimeApi; @@ -359,7 +361,7 @@ pub fn new_full_base( spawn_handle: task_manager.spawn_handle(), import_queue, block_announce_validator_builder: None, - warp_sync: Some(warp_sync), + warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)), })?; if config.offchain_worker.enabled { diff --git a/client/informant/src/display.rs b/client/informant/src/display.rs index 3d585a9985134..4da46ebb65edb 100644 --- a/client/informant/src/display.rs +++ b/client/informant/src/display.rs @@ -100,6 +100,11 @@ impl InformantDisplay { _, Some(WarpSyncProgress { phase: WarpSyncPhase::DownloadingBlocks(n), .. }), ) => ("⏩", "Block history".into(), format!(", #{}", n)), + ( + _, + _, + Some(WarpSyncProgress { phase: WarpSyncPhase::AwaitingTargetBlock, .. }), + ) => ("⏩", "Waiting for pending target block".into(), "".into()), (_, _, Some(warp)) => ( "⏩", "Warping".into(), diff --git a/client/network/common/src/sync/warp.rs b/client/network/common/src/sync/warp.rs index c9b9037542388..020642fb48c82 100644 --- a/client/network/common/src/sync/warp.rs +++ b/client/network/common/src/sync/warp.rs @@ -15,9 +15,10 @@ // along with Substrate. If not, see . use codec::{Decode, Encode}; +use futures::channel::oneshot; pub use sp_finality_grandpa::{AuthorityList, SetId}; use sp_runtime::traits::{Block as BlockT, NumberFor}; -use std::fmt; +use std::{fmt, sync::Arc}; /// Scale-encoded warp sync proof response. pub struct EncodedProof(pub Vec); @@ -29,6 +30,16 @@ pub struct WarpProofRequest { pub begin: B::Hash, } +/// The different types of warp syncing. +pub enum WarpSyncParams { + /// Standard warp sync for the relay chain + WithProvider(Arc>), + /// Skip downloading proofs and wait for a header of the state that should be downloaded. + /// + /// It is expected that the header provider ensures that the header is trusted. + WaitForTarget(oneshot::Receiver<::Header>), +} + /// Proof verification result. pub enum VerificationResult { /// Proof is valid, but the target was not reached. @@ -62,6 +73,8 @@ pub trait WarpSyncProvider: Send + Sync { pub enum WarpSyncPhase { /// Waiting for peers to connect. AwaitingPeers, + /// Waiting for target block to be received. + AwaitingTargetBlock, /// Downloading and verifying grandpa warp proofs. DownloadingWarpProofs, /// Downloading target block. @@ -78,6 +91,7 @@ impl fmt::Display for WarpSyncPhase { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { Self::AwaitingPeers => write!(f, "Waiting for peers"), + Self::AwaitingTargetBlock => write!(f, "Waiting for target block to be received"), Self::DownloadingWarpProofs => write!(f, "Downloading finality proofs"), Self::DownloadingTargetBlock => write!(f, "Downloading target block"), Self::DownloadingState => write!(f, "Downloading state"), diff --git a/client/network/sync/src/lib.rs b/client/network/sync/src/lib.rs index 312fc6f5b7947..ffc0edaf31e72 100644 --- a/client/network/sync/src/lib.rs +++ b/client/network/sync/src/lib.rs @@ -71,7 +71,7 @@ use sc_network_common::{ BlockAnnounce, BlockAnnouncesHandshake, BlockAttributes, BlockData, BlockRequest, BlockResponse, Direction, FromBlock, }, - warp::{EncodedProof, WarpProofRequest, WarpSyncPhase, WarpSyncProgress, WarpSyncProvider}, + warp::{EncodedProof, WarpProofRequest, WarpSyncParams, WarpSyncPhase, WarpSyncProgress}, BadPeer, ChainSync as ChainSyncT, ImportResult, Metrics, OnBlockData, OnBlockJustification, OnStateData, OpaqueBlockRequest, OpaqueBlockResponse, OpaqueStateRequest, OpaqueStateResponse, PeerInfo, PeerRequest, PollBlockAnnounceValidation, SyncMode, @@ -318,8 +318,10 @@ pub struct ChainSync { state_sync: Option>, /// Warp sync in progress, if any. warp_sync: Option>, - /// Warp sync provider. - warp_sync_provider: Option>>, + /// Warp sync params. + /// + /// Will be `None` after `self.warp_sync` is `Some(_)`. + warp_sync_params: Option>, /// Enable importing existing blocks. This is used used after the state download to /// catch up to the latest state while re-importing blocks. import_existing: bool, @@ -565,6 +567,7 @@ where info!("💔 New peer with unknown genesis hash {} ({}).", best_hash, best_number); return Err(BadPeer(who, rep::GENESIS_MISMATCH)) } + // If there are more than `MAJOR_SYNC_BLOCKS` in the import queue then we have // enough to do in the import queue that it's not worth kicking off // an ancestor search, which is what we do in the next match case below. @@ -630,17 +633,15 @@ where }, ); - if let SyncMode::Warp = &self.mode { + if let SyncMode::Warp = self.mode { if self.peers.len() >= MIN_PEERS_TO_START_WARP_SYNC && self.warp_sync.is_none() { log::debug!(target: "sync", "Starting warp state sync."); - if let Some(provider) = &self.warp_sync_provider { - self.warp_sync = - Some(WarpSync::new(self.client.clone(), provider.clone())); + if let Some(params) = self.warp_sync_params.take() { + self.warp_sync = Some(WarpSync::new(self.client.clone(), params)); } } } - Ok(req) }, Ok(BlockStatus::Queued) | @@ -1359,6 +1360,13 @@ where }, } } + + // Should be called before `process_outbound_requests` to ensure + // that a potential target block is directly leading to requests. + if let Some(warp_sync) = &mut self.warp_sync { + let _ = warp_sync.poll(cx); + } + self.process_outbound_requests(); while let Poll::Ready(result) = self.poll_pending_responses(cx) { @@ -1427,7 +1435,7 @@ where roles: Roles, block_announce_validator: Box + Send>, max_parallel_downloads: u32, - warp_sync_provider: Option>>, + warp_sync_params: Option>, metrics_registry: Option<&Registry>, network_service: service::network::NetworkServiceHandle, import_queue: Box>, @@ -1467,13 +1475,13 @@ where block_announce_validation_per_peer_stats: Default::default(), state_sync: None, warp_sync: None, - warp_sync_provider, import_existing: false, gap_sync: None, service_rx, network_service, block_request_protocol_name, state_request_protocol_name, + warp_sync_params, warp_sync_protocol_name, block_announce_protocol_name: block_announce_config .notifications_protocol diff --git a/client/network/sync/src/warp.rs b/client/network/sync/src/warp.rs index ab8a7c66b9856..bb64ff6a15890 100644 --- a/client/network/sync/src/warp.rs +++ b/client/network/sync/src/warp.rs @@ -19,24 +19,35 @@ //! Warp sync support. use crate::{ + oneshot, schema::v1::{StateRequest, StateResponse}, state::{ImportResult, StateSync}, }; +use futures::FutureExt; +use log::error; use sc_client_api::ProofProvider; use sc_network_common::sync::{ message::{BlockAttributes, BlockData, BlockRequest, Direction, FromBlock}, warp::{ - EncodedProof, VerificationResult, WarpProofRequest, WarpSyncPhase, WarpSyncProgress, - WarpSyncProvider, + EncodedProof, VerificationResult, WarpProofRequest, WarpSyncParams, WarpSyncPhase, + WarpSyncProgress, WarpSyncProvider, }, }; use sp_blockchain::HeaderBackend; use sp_finality_grandpa::{AuthorityList, SetId}; use sp_runtime::traits::{Block as BlockT, Header, NumberFor, Zero}; -use std::sync::Arc; +use std::{sync::Arc, task::Poll}; enum Phase { - WarpProof { set_id: SetId, authorities: AuthorityList, last_hash: B::Hash }, + WarpProof { + set_id: SetId, + authorities: AuthorityList, + last_hash: B::Hash, + warp_sync_provider: Arc>, + }, + PendingTargetBlock { + target_block: Option>, + }, TargetBlock(B::Header), State(StateSync), } @@ -61,7 +72,6 @@ pub enum TargetBlockImportResult { pub struct WarpSync { phase: Phase, client: Arc, - warp_sync_provider: Arc>, total_proof_bytes: u64, } @@ -70,21 +80,56 @@ where B: BlockT, Client: HeaderBackend + ProofProvider + 'static, { - /// Create a new instance. - pub fn new(client: Arc, warp_sync_provider: Arc>) -> Self { + /// Create a new instance. When passing a warp sync provider we will be checking for proof and + /// authorities. Alternatively we can pass a target block when we want to skip downloading + /// proofs, in this case we will continue polling until the target block is known. + pub fn new(client: Arc, warp_sync_params: WarpSyncParams) -> Self { let last_hash = client.hash(Zero::zero()).unwrap().expect("Genesis header always exists"); - let phase = Phase::WarpProof { - set_id: 0, - authorities: warp_sync_provider.current_authorities(), - last_hash, + match warp_sync_params { + WarpSyncParams::WithProvider(warp_sync_provider) => { + let phase = Phase::WarpProof { + set_id: 0, + authorities: warp_sync_provider.current_authorities(), + last_hash, + warp_sync_provider: warp_sync_provider.clone(), + }; + Self { client, phase, total_proof_bytes: 0 } + }, + WarpSyncParams::WaitForTarget(block) => Self { + client, + phase: Phase::PendingTargetBlock { target_block: Some(block) }, + total_proof_bytes: 0, + }, + } + } + + /// Poll to make progress. + /// + /// This only makes progress when `phase = Phase::PendingTargetBlock` and the pending block was + /// sent. + pub fn poll(&mut self, cx: &mut std::task::Context) { + let new_phase = if let Phase::PendingTargetBlock { target_block: Some(target_block) } = + &mut self.phase + { + match target_block.poll_unpin(cx) { + Poll::Ready(Ok(target)) => Phase::TargetBlock(target), + Poll::Ready(Err(e)) => { + error!(target: "sync", "Failed to get target block. Error: {:?}",e); + Phase::PendingTargetBlock { target_block: None } + }, + _ => return, + } + } else { + return }; - Self { client, warp_sync_provider, phase, total_proof_bytes: 0 } + + self.phase = new_phase; } /// Validate and import a state response. pub fn import_state(&mut self, response: StateResponse) -> ImportResult { match &mut self.phase { - Phase::WarpProof { .. } | Phase::TargetBlock(_) => { + Phase::WarpProof { .. } | Phase::TargetBlock(_) | Phase::PendingTargetBlock { .. } => { log::debug!(target: "sync", "Unexpected state response"); ImportResult::BadResponse }, @@ -95,12 +140,12 @@ where /// Validate and import a warp proof response. pub fn import_warp_proof(&mut self, response: EncodedProof) -> WarpProofImportResult { match &mut self.phase { - Phase::State(_) | Phase::TargetBlock(_) => { + Phase::State(_) | Phase::TargetBlock(_) | Phase::PendingTargetBlock { .. } => { log::debug!(target: "sync", "Unexpected warp proof response"); WarpProofImportResult::BadResponse }, - Phase::WarpProof { set_id, authorities, last_hash } => { - match self.warp_sync_provider.verify(&response, *set_id, authorities.clone()) { + Phase::WarpProof { set_id, authorities, last_hash, warp_sync_provider } => + match warp_sync_provider.verify(&response, *set_id, authorities.clone()) { Err(e) => { log::debug!(target: "sync", "Bad warp proof response: {}", e); WarpProofImportResult::BadResponse @@ -119,15 +164,14 @@ where self.phase = Phase::TargetBlock(header); WarpProofImportResult::Success }, - } - }, + }, } } /// Import the target block body. pub fn import_target_block(&mut self, block: BlockData) -> TargetBlockImportResult { match &mut self.phase { - Phase::WarpProof { .. } | Phase::State(_) => { + Phase::WarpProof { .. } | Phase::State(_) | Phase::PendingTargetBlock { .. } => { log::debug!(target: "sync", "Unexpected target block response"); TargetBlockImportResult::BadResponse }, @@ -168,8 +212,8 @@ where /// Produce next state request. pub fn next_state_request(&self) -> Option { match &self.phase { - Phase::WarpProof { .. } => None, - Phase::TargetBlock(_) => None, + Phase::WarpProof { .. } | Phase::TargetBlock(_) | Phase::PendingTargetBlock { .. } => + None, Phase::State(sync) => Some(sync.next_request()), } } @@ -178,15 +222,14 @@ where pub fn next_warp_proof_request(&self) -> Option> { match &self.phase { Phase::WarpProof { last_hash, .. } => Some(WarpProofRequest { begin: *last_hash }), - Phase::TargetBlock(_) => None, - Phase::State(_) => None, + Phase::TargetBlock(_) | Phase::State(_) | Phase::PendingTargetBlock { .. } => None, } } /// Produce next target block request. pub fn next_target_block_request(&self) -> Option<(NumberFor, BlockRequest)> { match &self.phase { - Phase::WarpProof { .. } => None, + Phase::WarpProof { .. } | Phase::State(_) | Phase::PendingTargetBlock { .. } => None, Phase::TargetBlock(header) => { let request = BlockRequest:: { id: 0, @@ -198,15 +241,14 @@ where }; Some((*header.number(), request)) }, - Phase::State(_) => None, } } /// Return target block hash if it is known. pub fn target_block_hash(&self) -> Option { match &self.phase { - Phase::WarpProof { .. } => None, - Phase::TargetBlock(_) => None, + Phase::WarpProof { .. } | Phase::TargetBlock(_) | Phase::PendingTargetBlock { .. } => + None, Phase::State(s) => Some(s.target()), } } @@ -214,7 +256,7 @@ where /// Return target block number if it is known. pub fn target_block_number(&self) -> Option> { match &self.phase { - Phase::WarpProof { .. } => None, + Phase::WarpProof { .. } | Phase::PendingTargetBlock { .. } => None, Phase::TargetBlock(header) => Some(*header.number()), Phase::State(s) => Some(s.target_block_num()), } @@ -223,8 +265,8 @@ where /// Check if the state is complete. pub fn is_complete(&self) -> bool { match &self.phase { - Phase::WarpProof { .. } => false, - Phase::TargetBlock(_) => false, + Phase::WarpProof { .. } | Phase::TargetBlock(_) | Phase::PendingTargetBlock { .. } => + false, Phase::State(sync) => sync.is_complete(), } } @@ -240,6 +282,10 @@ where phase: WarpSyncPhase::DownloadingTargetBlock, total_bytes: self.total_proof_bytes, }, + Phase::PendingTargetBlock { .. } => WarpSyncProgress { + phase: WarpSyncPhase::AwaitingTargetBlock, + total_bytes: self.total_proof_bytes, + }, Phase::State(sync) => WarpSyncProgress { phase: if self.is_complete() { WarpSyncPhase::ImportingState diff --git a/client/network/test/src/lib.rs b/client/network/test/src/lib.rs index b3653ac7c0f88..ccaebc976135b 100644 --- a/client/network/test/src/lib.rs +++ b/client/network/test/src/lib.rs @@ -31,7 +31,7 @@ use std::{ time::Duration, }; -use futures::{future::BoxFuture, prelude::*}; +use futures::{channel::oneshot, future::BoxFuture, prelude::*}; use libp2p::{build_multiaddr, PeerId}; use log::trace; use parking_lot::Mutex; @@ -56,7 +56,9 @@ use sc_network_common::{ }, protocol::{role::Roles, ProtocolName}, service::{NetworkBlock, NetworkStateInfo, NetworkSyncForkRequest}, - sync::warp::{AuthorityList, EncodedProof, SetId, VerificationResult, WarpSyncProvider}, + sync::warp::{ + AuthorityList, EncodedProof, SetId, VerificationResult, WarpSyncParams, WarpSyncProvider, + }, }; use sc_network_light::light_client_requests::handler::LightClientRequestHandler; use sc_network_sync::{ @@ -722,6 +724,8 @@ pub struct FullPeerConfig { pub extra_storage: Option, /// Enable transaction indexing. pub storage_chain: bool, + /// Optional target block header to sync to + pub target_block: Option<::Header>, } #[async_trait::async_trait] @@ -867,6 +871,15 @@ where let warp_sync = Arc::new(TestWarpSyncProvider(client.clone())); + let warp_sync_params = match config.target_block { + Some(target_block) => { + let (sender, receiver) = oneshot::channel::<::Header>(); + let _ = sender.send(target_block); + WarpSyncParams::WaitForTarget(receiver) + }, + _ => WarpSyncParams::WithProvider(warp_sync.clone()), + }; + let warp_protocol_config = { let (handler, protocol_config) = warp_request_handler::RequestHandler::new( protocol_id.clone(), @@ -887,6 +900,7 @@ where .unwrap_or_else(|| Box::new(DefaultBlockAnnounceValidator)); let (chain_sync_network_provider, chain_sync_network_handle) = NetworkServiceProvider::new(); + let (chain_sync, chain_sync_service, block_announce_config) = ChainSync::new( match network_config.sync_mode { SyncMode::Full => sc_network_common::sync::SyncMode::Full, @@ -903,7 +917,7 @@ where Roles::from(if config.is_authority { &Role::Authority } else { &Role::Full }), block_announce_validator, network_config.max_parallel_downloads, - Some(warp_sync), + Some(warp_sync_params), None, chain_sync_network_handle, import_queue.service(), diff --git a/client/network/test/src/sync.rs b/client/network/test/src/sync.rs index b629574fe9874..8c230640355c9 100644 --- a/client/network/test/src/sync.rs +++ b/client/network/test/src/sync.rs @@ -1249,6 +1249,44 @@ async fn warp_sync() { .await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn warp_sync_to_target_block() { + sp_tracing::try_init_simple(); + let mut net = TestNet::new(0); + // Create 3 synced peers and 1 peer trying to warp sync. + net.add_full_peer_with_config(Default::default()); + net.add_full_peer_with_config(Default::default()); + net.add_full_peer_with_config(Default::default()); + + let blocks = net.peer(0).push_blocks(64, false); + let target = blocks[63]; + net.peer(1).push_blocks(64, false); + net.peer(2).push_blocks(64, false); + + let target_block = net.peer(0).client.header(target).unwrap().unwrap(); + + net.add_full_peer_with_config(FullPeerConfig { + sync_mode: SyncMode::Warp, + target_block: Some(target_block), + ..Default::default() + }); + + net.run_until_sync().await; + assert!(net.peer(3).client().has_state_at(&BlockId::Number(64))); + + // Wait for peer 1 download block history + futures::future::poll_fn::<(), _>(|cx| { + net.poll(cx); + let peer = net.peer(3); + if blocks.iter().all(|b| peer.has_body(*b)) { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + .await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn syncs_huge_blocks() { use sp_core::storage::well_known_keys::HEAP_PAGES; diff --git a/client/service/src/builder.rs b/client/service/src/builder.rs index 0b09f550ce338..a737601f71b83 100644 --- a/client/service/src/builder.rs +++ b/client/service/src/builder.rs @@ -43,7 +43,7 @@ use sc_network_bitswap::BitswapRequestHandler; use sc_network_common::{ protocol::role::Roles, service::{NetworkStateInfo, NetworkStatusProvider}, - sync::warp::WarpSyncProvider, + sync::warp::WarpSyncParams, }; use sc_network_light::light_client_requests::handler::LightClientRequestHandler; use sc_network_sync::{ @@ -759,8 +759,8 @@ pub struct BuildNetworkParams<'a, TBl: BlockT, TExPool, TImpQu, TCl> { /// A block announce validator builder. pub block_announce_validator_builder: Option) -> Box + Send> + Send>>, - /// An optional warp sync provider. - pub warp_sync: Option>>, + /// Optional warp sync params. + pub warp_sync_params: Option>, } /// Build the network service, the network status sinks and an RPC sender. pub fn build_network( @@ -795,12 +795,12 @@ where spawn_handle, import_queue, block_announce_validator_builder, - warp_sync, + warp_sync_params, } = params; let mut request_response_protocol_configs = Vec::new(); - if warp_sync.is_none() && config.network.sync_mode.is_warp() { + if warp_sync_params.is_none() && config.network.sync_mode.is_warp() { return Err("Warp sync enabled, but no warp sync provider configured.".into()) } @@ -845,8 +845,8 @@ where protocol_config }; - let (warp_sync_provider, warp_sync_protocol_config) = warp_sync - .map(|provider| { + let warp_sync_protocol_config = match warp_sync_params.as_ref() { + Some(WarpSyncParams::WithProvider(warp_with_provider)) => { // Allow both outgoing and incoming requests. let (handler, protocol_config) = WarpSyncRequestHandler::new( protocol_id.clone(), @@ -856,12 +856,13 @@ where .flatten() .expect("Genesis block exists; qed"), config.chain_spec.fork_id(), - provider.clone(), + warp_with_provider.clone(), ); spawn_handle.spawn("warp-sync-request-handler", Some("networking"), handler.run()); - (Some(provider), Some(protocol_config)) - }) - .unwrap_or_default(); + Some(protocol_config) + }, + _ => None, + }; let light_client_request_protocol_config = { // Allow both outgoing and incoming requests. @@ -888,7 +889,7 @@ where Roles::from(&config.role), block_announce_validator, config.network.max_parallel_downloads, - warp_sync_provider, + warp_sync_params, config.prometheus_config.as_ref().map(|config| config.registry.clone()).as_ref(), chain_sync_network_handle, import_queue.service(), diff --git a/client/service/src/lib.rs b/client/service/src/lib.rs index 1529b822ade32..e3dcf012892c3 100644 --- a/client/service/src/lib.rs +++ b/client/service/src/lib.rs @@ -73,6 +73,7 @@ pub use sc_chain_spec::{ pub use sc_consensus::ImportQueue; pub use sc_executor::NativeExecutionDispatch; +pub use sc_network_common::sync::warp::WarpSyncParams; #[doc(hidden)] pub use sc_network_transactions::config::{TransactionImport, TransactionImportFuture}; pub use sc_rpc::{