diff --git a/client/network/common/src/sync.rs b/client/network/common/src/sync.rs index 2ee8f8c51814b..020b2c9efa4c7 100644 --- a/client/network/common/src/sync.rs +++ b/client/network/common/src/sync.rs @@ -96,6 +96,8 @@ pub enum OnBlockData { Import(BlockOrigin, Vec>), /// A new block request needs to be made to the given peer. Request(PeerId, BlockRequest), + /// Continue processing events. + Continue, } /// Result of [`ChainSync::on_block_justification`]. diff --git a/client/network/common/src/sync/warp.rs b/client/network/common/src/sync/warp.rs index 339a4c33a7eeb..c9b9037542388 100644 --- a/client/network/common/src/sync/warp.rs +++ b/client/network/common/src/sync/warp.rs @@ -64,6 +64,8 @@ pub enum WarpSyncPhase { AwaitingPeers, /// Downloading and verifying grandpa warp proofs. DownloadingWarpProofs, + /// Downloading target block. + DownloadingTargetBlock, /// Downloading state data. DownloadingState, /// Importing state. @@ -77,6 +79,7 @@ impl fmt::Display for WarpSyncPhase { match self { Self::AwaitingPeers => write!(f, "Waiting for peers"), Self::DownloadingWarpProofs => write!(f, "Downloading finality proofs"), + Self::DownloadingTargetBlock => write!(f, "Downloading target block"), Self::DownloadingState => write!(f, "Downloading state"), Self::ImportingState => write!(f, "Importing state"), Self::DownloadingBlocks(n) => write!(f, "Downloading block history (#{})", n), diff --git a/client/network/src/protocol.rs b/client/network/src/protocol.rs index 9bcf363cd0692..a7af43d53d3b7 100644 --- a/client/network/src/protocol.rs +++ b/client/network/src/protocol.rs @@ -627,6 +627,7 @@ where CustomMessageOutcome::BlockImport(origin, blocks), Ok(OnBlockData::Request(peer, req)) => prepare_block_request(self.chain_sync.as_ref(), &mut self.peers, peer, req), + Ok(OnBlockData::Continue) => CustomMessageOutcome::None, Err(BadPeer(id, repu)) => { self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC); self.peerset_handle.report_peer(id, repu); @@ -974,6 +975,7 @@ where CustomMessageOutcome::BlockImport(origin, blocks), Ok(OnBlockData::Request(peer, req)) => prepare_block_request(self.chain_sync.as_ref(), &mut self.peers, peer, req), + Ok(OnBlockData::Continue) => CustomMessageOutcome::None, Err(BadPeer(id, repu)) => { self.behaviour.disconnect_peer(&id, HARDCODED_PEERSETS_SYNC); self.peerset_handle.report_peer(id, repu); diff --git a/client/network/sync/src/lib.rs b/client/network/sync/src/lib.rs index aae5f4de353fe..6ad4a8fbbdcc5 100644 --- a/client/network/sync/src/lib.rs +++ b/client/network/sync/src/lib.rs @@ -80,6 +80,7 @@ use std::{ pin::Pin, sync::Arc, }; +use warp::TargetBlockImportResult; mod extra_requests; @@ -315,6 +316,8 @@ pub enum PeerSyncState { DownloadingState, /// Downloading warp proof. DownloadingWarpProof, + /// Downloading warp sync target block. + DownloadingWarpTargetBlock, /// Actively downloading block history after warp sync. DownloadingGap(NumberFor), } @@ -659,10 +662,11 @@ where } fn block_requests(&mut self) -> Box)> + '_> { - if self.allowed_requests.is_empty() || - self.state_sync.is_some() || - self.mode == SyncMode::Warp - { + if self.mode == SyncMode::Warp { + return Box::new(std::iter::once(self.warp_target_block_request()).flatten()) + } + + if self.allowed_requests.is_empty() || self.state_sync.is_some() { return Box::new(std::iter::empty()) } @@ -824,7 +828,7 @@ where // Only one pending state request is allowed. return None } - if let Some(request) = sync.next_warp_poof_request() { + if let Some(request) = sync.next_warp_proof_request() { let mut targets: Vec<_> = self.peers.values().map(|p| p.best_number).collect(); if !targets.is_empty() { targets.sort(); @@ -1031,6 +1035,40 @@ where Vec::new() } }, + PeerSyncState::DownloadingWarpTargetBlock => { + peer.state = PeerSyncState::Available; + if let Some(warp_sync) = &mut self.warp_sync { + if blocks.len() == 1 { + validate_blocks::(&blocks, who, Some(request))?; + match warp_sync.import_target_block( + blocks.pop().expect("`blocks` len checked above."), + ) { + TargetBlockImportResult::Success => + return Ok(OnBlockData::Continue), + TargetBlockImportResult::BadResponse => + return Err(BadPeer(*who, rep::VERIFICATION_FAIL)), + } + } else if blocks.is_empty() { + debug!(target: "sync", "Empty block response from {}", who); + return Err(BadPeer(*who, rep::NO_BLOCK)) + } else { + debug!( + target: "sync", + "Too many blocks ({}) in warp target block response from {}", + blocks.len(), + who, + ); + return Err(BadPeer(*who, rep::NOT_REQUESTED)) + } + } else { + debug!( + target: "sync", + "Logic error: we think we are downloading warp target block from {}, but no warp sync is happening.", + who, + ); + return Ok(OnBlockData::Continue) + } + }, PeerSyncState::Available | PeerSyncState::DownloadingJustification(..) | PeerSyncState::DownloadingState | @@ -1112,14 +1150,14 @@ where }; match import_result { - state::ImportResult::Import(hash, header, state) => { + state::ImportResult::Import(hash, header, state, body, justifications) => { let origin = BlockOrigin::NetworkInitialSync; let block = IncomingBlock { hash, header: Some(header), - body: None, + body, indexed_body: None, - justifications: None, + justifications, origin: None, allow_missing_state: true, import_existing: true, @@ -1399,8 +1437,13 @@ where number, hash, ); - self.state_sync = - Some(StateSync::new(self.client.clone(), header, *skip_proofs)); + self.state_sync = Some(StateSync::new( + self.client.clone(), + header, + None, + None, + *skip_proofs, + )); self.allowed_requests.set_all(); } } @@ -2163,6 +2206,33 @@ where }) .collect() } + + /// Generate block request for downloading of the target block body during warp sync. + fn warp_target_block_request(&mut self) -> Option<(&PeerId, BlockRequest)> { + if let Some(sync) = &self.warp_sync { + if self.allowed_requests.is_empty() || + sync.is_complete() || + self.peers + .iter() + .any(|(_, peer)| peer.state == PeerSyncState::DownloadingWarpTargetBlock) + { + // Only one pending warp target block request is allowed. + return None + } + if let Some((target_number, request)) = sync.next_target_block_request() { + // Find a random peer that has a block with the target number. + for (id, peer) in self.peers.iter_mut() { + if peer.state.is_available() && peer.best_number >= target_number { + trace!(target: "sync", "New warp target block request for {}", id); + peer.state = PeerSyncState::DownloadingWarpTargetBlock; + self.allowed_requests.clear(); + return Some((id, request)) + } + } + } + } + None + } } // This is purely during a backwards compatible transitionary period and should be removed diff --git a/client/network/sync/src/state.rs b/client/network/sync/src/state.rs index e70d3b6b33a28..9f64b52334c8a 100644 --- a/client/network/sync/src/state.rs +++ b/client/network/sync/src/state.rs @@ -26,7 +26,10 @@ use sc_consensus::ImportedState; use sc_network_common::sync::StateDownloadProgress; use smallvec::SmallVec; use sp_core::storage::well_known_keys; -use sp_runtime::traits::{Block as BlockT, Header, NumberFor}; +use sp_runtime::{ + traits::{Block as BlockT, Header, NumberFor}, + Justifications, +}; use std::{collections::HashMap, sync::Arc}; /// State sync state machine. Accumulates partial state data until it @@ -35,6 +38,8 @@ pub struct StateSync { target_block: B::Hash, target_header: B::Header, target_root: B::Hash, + target_body: Option>, + target_justifications: Option, last_key: SmallVec<[Vec; 2]>, state: HashMap, (Vec<(Vec, Vec)>, Vec>)>, complete: bool, @@ -46,7 +51,7 @@ pub struct StateSync { /// Import state chunk result. pub enum ImportResult { /// State is complete and ready for import. - Import(B::Hash, B::Header, ImportedState), + Import(B::Hash, B::Header, ImportedState, Option>, Option), /// Continue downloading. Continue, /// Bad state chunk. @@ -59,12 +64,20 @@ where Client: ProofProvider + Send + Sync + 'static, { /// Create a new instance. - pub fn new(client: Arc, target: B::Header, skip_proof: bool) -> Self { + pub fn new( + client: Arc, + target_header: B::Header, + target_body: Option>, + target_justifications: Option, + skip_proof: bool, + ) -> Self { Self { client, - target_block: target.hash(), - target_root: *target.state_root(), - target_header: target, + target_block: target_header.hash(), + target_root: *target_header.state_root(), + target_header, + target_body, + target_justifications, last_key: SmallVec::default(), state: HashMap::default(), complete: false, @@ -213,6 +226,8 @@ where block: self.target_block, state: std::mem::take(&mut self.state).into(), }, + self.target_body.clone(), + self.target_justifications.clone(), ) } else { ImportResult::Continue diff --git a/client/network/sync/src/warp.rs b/client/network/sync/src/warp.rs index f3fad6c1b7fdb..4f2a71d98613e 100644 --- a/client/network/sync/src/warp.rs +++ b/client/network/sync/src/warp.rs @@ -23,17 +23,21 @@ use crate::{ state::{ImportResult, StateSync}, }; use sc_client_api::ProofProvider; -use sc_network_common::sync::warp::{ - EncodedProof, VerificationResult, WarpProofRequest, WarpSyncPhase, WarpSyncProgress, - WarpSyncProvider, +use sc_network_common::sync::{ + message::{BlockAttributes, BlockData, BlockRequest, Direction, FromBlock}, + warp::{ + EncodedProof, VerificationResult, WarpProofRequest, WarpSyncPhase, WarpSyncProgress, + WarpSyncProvider, + }, }; use sp_blockchain::HeaderBackend; use sp_finality_grandpa::{AuthorityList, SetId}; -use sp_runtime::traits::{Block as BlockT, NumberFor, Zero}; +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 }, + TargetBlock(B::Header), State(StateSync), } @@ -45,6 +49,14 @@ pub enum WarpProofImportResult { BadResponse, } +/// Import target block result. +pub enum TargetBlockImportResult { + /// Import was successful. + Success, + /// Invalid block. + BadResponse, +} + /// Warp sync state machine. Accumulates warp proofs and state. pub struct WarpSync { phase: Phase, @@ -72,7 +84,7 @@ where /// Validate and import a state response. pub fn import_state(&mut self, response: StateResponse) -> ImportResult { match &mut self.phase { - Phase::WarpProof { .. } => { + Phase::WarpProof { .. } | Phase::TargetBlock(_) => { log::debug!(target: "sync", "Unexpected state response"); ImportResult::BadResponse }, @@ -83,7 +95,7 @@ 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::State(_) | Phase::TargetBlock(_) => { log::debug!(target: "sync", "Unexpected warp proof response"); WarpProofImportResult::BadResponse }, @@ -104,8 +116,7 @@ where Ok(VerificationResult::Complete(new_set_id, _, header)) => { log::debug!(target: "sync", "Verified complete proof, set_id={:?}", new_set_id); self.total_proof_bytes += response.0.len() as u64; - let state_sync = StateSync::new(self.client.clone(), header, false); - self.phase = Phase::State(state_sync); + self.phase = Phase::TargetBlock(header); WarpProofImportResult::Success }, } @@ -113,35 +124,100 @@ where } } + /// Import the target block body. + pub fn import_target_block(&mut self, block: BlockData) -> TargetBlockImportResult { + match &mut self.phase { + Phase::WarpProof { .. } | Phase::State(_) => { + log::debug!(target: "sync", "Unexpected target block response"); + TargetBlockImportResult::BadResponse + }, + Phase::TargetBlock(header) => + if let Some(block_header) = &block.header { + if block_header == header { + if block.body.is_some() { + let state_sync = StateSync::new( + self.client.clone(), + header.clone(), + block.body, + block.justifications, + false, + ); + self.phase = Phase::State(state_sync); + TargetBlockImportResult::Success + } else { + log::debug!( + target: "sync", + "Importing target block failed: missing body.", + ); + TargetBlockImportResult::BadResponse + } + } else { + log::debug!( + target: "sync", + "Importing target block failed: different header.", + ); + TargetBlockImportResult::BadResponse + } + } else { + log::debug!(target: "sync", "Importing target block failed: missing header."); + TargetBlockImportResult::BadResponse + }, + } + } + /// Produce next state request. pub fn next_state_request(&self) -> Option { match &self.phase { Phase::WarpProof { .. } => None, + Phase::TargetBlock(_) => None, Phase::State(sync) => Some(sync.next_request()), } } /// Produce next warp proof request. - pub fn next_warp_poof_request(&self) -> Option> { + pub fn next_warp_proof_request(&self) -> Option> { match &self.phase { - Phase::State(_) => None, Phase::WarpProof { last_hash, .. } => Some(WarpProofRequest { begin: *last_hash }), + Phase::TargetBlock(_) => None, + Phase::State(_) => None, + } + } + + /// Produce next target block request. + pub fn next_target_block_request(&self) -> Option<(NumberFor, BlockRequest)> { + match &self.phase { + Phase::WarpProof { .. } => None, + Phase::TargetBlock(header) => { + let request = BlockRequest:: { + id: 0, + fields: BlockAttributes::HEADER | + BlockAttributes::BODY | BlockAttributes::JUSTIFICATION, + from: FromBlock::Hash(header.hash()), + to: Some(header.hash()), + direction: Direction::Ascending, + max: Some(1), + }; + 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::State(s) => Some(s.target()), Phase::WarpProof { .. } => None, + Phase::TargetBlock(_) => None, + Phase::State(s) => Some(s.target()), } } /// Return target block number if it is known. pub fn target_block_number(&self) -> Option> { match &self.phase { - Phase::State(s) => Some(s.target_block_num()), Phase::WarpProof { .. } => None, + Phase::TargetBlock(header) => Some(*header.number()), + Phase::State(s) => Some(s.target_block_num()), } } @@ -149,6 +225,7 @@ where pub fn is_complete(&self) -> bool { match &self.phase { Phase::WarpProof { .. } => false, + Phase::TargetBlock(_) => false, Phase::State(sync) => sync.is_complete(), } } @@ -160,6 +237,10 @@ where phase: WarpSyncPhase::DownloadingWarpProofs, total_bytes: self.total_proof_bytes, }, + Phase::TargetBlock(_) => WarpSyncProgress { + phase: WarpSyncPhase::DownloadingTargetBlock, + 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 5a29e587ceff5..e78b91a4e04ee 100644 --- a/client/network/test/src/lib.rs +++ b/client/network/test/src/lib.rs @@ -69,7 +69,7 @@ use sc_network_sync::{ use sc_service::client::Client; use sp_blockchain::{ well_known_cache_keys::{self, Id as CacheKeyId}, - HeaderBackend, Info as BlockchainInfo, Result as ClientResult, + Backend as BlockchainBackend, HeaderBackend, Info as BlockchainInfo, Result as ClientResult, }; use sp_consensus::{ block_validation::{BlockAnnounceValidator, DefaultBlockAnnounceValidator}, @@ -540,6 +540,13 @@ where .map(|backend| backend.blockchain().header(BlockId::hash(*hash)).unwrap().is_some()) .unwrap_or(false) } + + pub fn has_body(&self, hash: &H256) -> bool { + self.backend + .as_ref() + .map(|backend| backend.blockchain().body(BlockId::hash(*hash)).unwrap().is_some()) + .unwrap_or(false) + } } pub trait BlockImportAdapterFull: diff --git a/client/network/test/src/sync.rs b/client/network/test/src/sync.rs index c0778767b75af..6115e0e3b2c86 100644 --- a/client/network/test/src/sync.rs +++ b/client/network/test/src/sync.rs @@ -1192,7 +1192,7 @@ fn warp_sync() { ..Default::default() }); let gap_end = net.peer(0).push_blocks(63, false); - net.peer(0).push_blocks(1, false); + let target = net.peer(0).push_blocks(1, false); net.peer(1).push_blocks(64, false); net.peer(2).push_blocks(64, false); // Wait for peer 1 to sync state. @@ -1203,7 +1203,7 @@ fn warp_sync() { // Wait for peer 1 download block history block_on(futures::future::poll_fn::<(), _>(|cx| { net.poll(cx); - if net.peer(3).has_block(&gap_end) { + if net.peer(3).has_body(&gap_end) && net.peer(3).has_body(&target) { Poll::Ready(()) } else { Poll::Pending