pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-04-26 19:17:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

add warp to target block for parachains (#12761)

Browse Source 
* add warp to target block for parachains

* fix for failing tests

* format using  `Cargo +nightly fmt`

* Remove blocking based on PR comments and create new `WarpSync` on poll

* remove method from trait

* add tests for wait for target

* Update client/network/common/src/sync/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/common/src/sync/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/lib.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* code refactor based on pr comments

* Second round of PR comments

* Third round of pr comments

* add comments to explain logic

* Update client/network/sync/src/lib.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/lib.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/lib.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* code refactor based on last PR comments

* move warp sync polling before `process_outbound_requests`

Add error message if target block fails to be retreived

* Update client/network/sync/src/warp.rs

Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>

* Update client/network/sync/src/lib.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* fmt after code suggestions

* rebase changes

* Bring down the node if the target block fails to return

* Revert "Bring down the node if the target block fails to return"

This reverts commit c0ecb220d66dd8e7b1a5ee29831b776f4f18d024.

* Update client/network/common/src/sync/warp.rs

Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com>

* Update client/network/common/src/sync/warp.rs

Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com>

* use matching on polling to avoid calling poll more than once

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/sync/src/warp.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* fix typo on comment

* update snapshot with new folder structure

* Upload snapshot

* Bump zombienet

* bump zombienet again

* Improve test

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* Update client/network/test/src/sync.rs

Co-authored-by: Bastian Köcher <git@kchr.de>

* fix tests

* dummy commit to restart builds

* Converted the target block to an optional value that is set to `None` when an error occurs

* dummy commit to restart builds

---------

Co-authored-by: Bastian Köcher <git@kchr.de>
Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>
Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com>
Co-authored-by: Sebastian Kunert <skunert49@gmail.com>
This commit is contained in:
Sam Elamin
2023-02-14 17:46:51 +00:00
committed by GitHub
parent 4af64eb071
commit df24729d74
10 changed files with 189 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/network/sync/src/lib.rs
+18 -10
View File
@@ -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<B: BlockT, Client> {
state_sync: Option<StateSync<B, Client>>,
/// Warp sync in progress, if any.
warp_sync: Option<WarpSync<B, Client>>,
/// Warp sync provider.
warp_sync_provider: Option<Arc<dyn WarpSyncProvider<B>>>,
/// Warp sync params.
///
/// Will be `None` after `self.warp_sync` is `Some(_)`.
warp_sync_params: Option<WarpSyncParams<B>>,
/// 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<dyn BlockAnnounceValidator<B> + Send>,
max_parallel_downloads: u32,
warp_sync_provider: Option<Arc<dyn WarpSyncProvider<B>>>,
warp_sync_params: Option<WarpSyncParams<B>>,
metrics_registry: Option<&Registry>,
network_service: service::network::NetworkServiceHandle,
import_queue: Box<dyn ImportQueueService<B>>,
@@ -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
substrate/client/network/sync/src/warp.rs
+76 -30
View File
@@ -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<B: BlockT, Client> {
WarpProof { set_id: SetId, authorities: AuthorityList, last_hash: B::Hash },
WarpProof {
set_id: SetId,
authorities: AuthorityList,
last_hash: B::Hash,
warp_sync_provider: Arc<dyn WarpSyncProvider<B>>,
},
PendingTargetBlock {
target_block: Option<oneshot::Receiver<B::Header>>,
},
TargetBlock(B::Header),
State(StateSync<B, Client>),
}
@@ -61,7 +72,6 @@ pub enum TargetBlockImportResult {
pub struct WarpSync<B: BlockT, Client> {
phase: Phase<B, Client>,
client: Arc<Client>,
warp_sync_provider: Arc<dyn WarpSyncProvider<B>>,
total_proof_bytes: u64,
}
@@ -70,21 +80,56 @@ where
B: BlockT,
Client: HeaderBackend<B> + ProofProvider<B> + 'static,
{
/// Create a new instance.
pub fn new(client: Arc<Client>, warp_sync_provider: Arc<dyn WarpSyncProvider<B>>) -> 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<Client>, warp_sync_params: WarpSyncParams<B>) -> 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<B> {
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<B>) -> 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<StateRequest> {
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<WarpProofRequest<B>> {
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<B>, BlockRequest<B>)> {
match &self.phase {
Phase::WarpProof { .. } => None,
Phase::WarpProof { .. } | Phase::State(_) | Phase::PendingTargetBlock { .. } => None,
Phase::TargetBlock(header) => {
let request = BlockRequest::<B> {
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<B::Hash> {
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<NumberFor<B>> {
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