diff --git a/bin/node-template/node/src/service.rs b/bin/node-template/node/src/service.rs index 687db46cf4ea8..ae6ad867583d8 100644 --- a/bin/node-template/node/src/service.rs +++ b/bin/node-template/node/src/service.rs @@ -6,8 +6,8 @@ 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_network_common::sync::warp::WarpSyncParams; +use sc_service::{error::Error as ServiceError, Configuration, TaskManager}; use sc_telemetry::{Telemetry, TelemetryWorker}; use sp_consensus_aura::sr25519::AuthorityPair as AuraPair; use std::{sync::Arc, time::Duration}; diff --git a/bin/node/cli/src/service.rs b/bin/node/cli/src/service.rs index 847ce371072a0..dcb9dc2ea3c20 100644 --- a/bin/node/cli/src/service.rs +++ b/bin/node/cli/src/service.rs @@ -30,7 +30,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, sync::warp::WarpSyncParams}; +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; diff --git a/client/network/common/src/sync/warp.rs b/client/network/common/src/sync/warp.rs index 445af3b67d7a9..90e3adcd0b49f 100644 --- a/client/network/common/src/sync/warp.rs +++ b/client/network/common/src/sync/warp.rs @@ -15,10 +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, sync::Arc}; -use futures::channel::oneshot; /// Scale-encoded warp sync proof response. pub struct EncodedProof(pub Vec); diff --git a/client/network/sync/Cargo.toml b/client/network/sync/Cargo.toml index f4c91717964c4..841388c7a68ee 100644 --- a/client/network/sync/Cargo.toml +++ b/client/network/sync/Cargo.toml @@ -19,7 +19,6 @@ prost-build = "0.11" [dependencies] array-bytes = "4.1" async-trait = "0.1.58" -async-std = { version = "1.11.0", default-features = false } codec = { package = "parity-scale-codec", version = "3.0.0", features = ["derive"] } futures = "0.3.21" libp2p = "0.49.0" diff --git a/client/network/sync/src/lib.rs b/client/network/sync/src/lib.rs index 7fbc6df8aa283..c4dbabaa6e3ce 100644 --- a/client/network/sync/src/lib.rs +++ b/client/network/sync/src/lib.rs @@ -533,6 +533,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. @@ -601,18 +602,25 @@ where if let SyncMode::Warp = &self.mode { if self.peers.len() >= MIN_PEERS_TO_START_WARP_SYNC && self.warp_sync.is_none() { - match self.warp_sync_params.as_mut().unwrap() { - WarpSyncParams::WithProvider(warp_with_provider) => { - log::debug!(target: "sync", "Starting warp state sync."); - self.warp_sync = Some(WarpSync::new(self.client.clone(), warp_with_provider.clone())); - } - WarpSyncParams::WaitForTarget(header) => { - log::debug!(target: "sync", "Waiting for target block."); - async_std::task::block_on(async { - self.warp_sync = Some(WarpSync::new_with_target_block(self.client.clone(), header.await.unwrap())); - }); - } - } + match self.warp_sync_params.as_mut() { + Some(WarpSyncParams::WithProvider(warp_with_provider)) => { + log::debug!(target: "sync", "Starting warp state sync."); + self.warp_sync = Some(WarpSync::new( + self.client.clone(), + warp_with_provider.clone(), + )); + }, + Some(WarpSyncParams::WaitForTarget(target_block)) => { + log::debug!(target: "sync", "Waiting for target block."); + futures::executor::block_on(async { + self.warp_sync = Some(WarpSync::new_with_target_block( + self.client.clone(), + target_block.await.unwrap(), + )); + }); + }, + None => {}, + } } } Ok(req) diff --git a/client/network/sync/src/warp.rs b/client/network/sync/src/warp.rs index 632fee9aa6b36..b998547a5a4b4 100644 --- a/client/network/sync/src/warp.rs +++ b/client/network/sync/src/warp.rs @@ -36,7 +36,12 @@ use sp_runtime::traits::{Block as BlockT, Header, NumberFor, Zero}; use std::sync::Arc; enum Phase { - WarpProof { set_id: SetId, authorities: AuthorityList, last_hash: B::Hash, warp_sync_provider: Arc> }, + WarpProof { + set_id: SetId, + authorities: AuthorityList, + last_hash: B::Hash, + warp_sync_provider: Arc>, + }, TargetBlock(B::Header), State(StateSync), } @@ -76,13 +81,15 @@ where set_id: 0, authorities: warp_sync_provider.current_authorities(), last_hash, - warp_sync_provider + warp_sync_provider, }; Self { client, phase, total_proof_bytes: 0 } } - pub fn new_with_target_block(client: Arc, target: ::Header) -> Self { - let phase = Phase::TargetBlock(target); + /// Create a new instance, skip the proof downloading and verification, directly go with a + /// target block + pub fn new_with_target_block(client: Arc, target_block: ::Header) -> Self { + let phase = Phase::TargetBlock(target_block); Self { client, phase, total_proof_bytes: 0 } } @@ -104,7 +111,7 @@ where log::debug!(target: "sync", "Unexpected warp proof response"); WarpProofImportResult::BadResponse }, - Phase::WarpProof { set_id, authorities, last_hash, warp_sync_provider } => { + 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); @@ -124,8 +131,7 @@ where self.phase = Phase::TargetBlock(header); WarpProofImportResult::Success }, - } - }, + }, } } diff --git a/client/network/test/src/lib.rs b/client/network/test/src/lib.rs index 9508889ef15ae..17ab5685b84c9 100644 --- a/client/network/test/src/lib.rs +++ b/client/network/test/src/lib.rs @@ -57,7 +57,9 @@ use sc_network_common::{ }, protocol::{role::Roles, ProtocolName}, service::{NetworkBlock, NetworkStateInfo, NetworkSyncForkRequest}, - sync::warp::{AuthorityList, EncodedProof, SetId, VerificationResult, WarpSyncParams, WarpSyncProvider}, + sync::warp::{ + AuthorityList, EncodedProof, SetId, VerificationResult, WarpSyncParams, WarpSyncProvider, + }, }; use sc_network_light::light_client_requests::handler::LightClientRequestHandler; use sc_network_sync::{ diff --git a/client/service/src/builder.rs b/client/service/src/builder.rs index 01578326353b7..e06999314088e 100644 --- a/client/service/src/builder.rs +++ b/client/service/src/builder.rs @@ -816,10 +816,9 @@ where protocol_config }; - let mut warp_sync_protocol_config = None; - match warp_sync_params.as_ref().unwrap() { - WarpSyncParams::WithProvider(warp_with_provider) => { - (_, warp_sync_protocol_config) = Some(warp_with_provider) + let warp_sync_protocol_config = match warp_sync_params.as_ref() { + Some(WarpSyncParams::WithProvider(warp_with_provider)) => { + let (_, warp_sync_protocol_config) = Some(warp_with_provider) .map(|provider| { // Allow both outgoing and incoming requests. let (handler, protocol_config) = WarpSyncRequestHandler::new( @@ -832,14 +831,18 @@ where config.chain_spec.fork_id(), provider.clone(), ); - spawn_handle.spawn("warp-sync-request-handler", Some("networking"), handler.run()); + spawn_handle.spawn( + "warp-sync-request-handler", + Some("networking"), + handler.run(), + ); (Some(provider), Some(protocol_config)) }) .unwrap_or_default(); + warp_sync_protocol_config }, - _ => { - } - } + _ => None, + }; let light_client_request_protocol_config = { // Allow both outgoing and incoming requests.