Allow import withouth state verification (#4031)

* Allow import without state verification

* Explicit None

Co-Authored-By: Robert Habermeier <rphmeier@gmail.com>
This commit is contained in:
Arkadiy Paronyan
2019-11-07 15:25:41 +01:00
committed by Gavin Wood
parent 021f3a3f06
commit cca9ab436c
20 changed files with 231 additions and 87 deletions
+45 -16
View File
@@ -457,6 +457,7 @@ pub struct BlockImportOperation<Block: BlockT, H: Hasher> {
aux_ops: Vec<(Vec<u8>, Option<Vec<u8>>)>, aux_ops: Vec<(Vec<u8>, Option<Vec<u8>>)>,
finalized_blocks: Vec<(BlockId<Block>, Option<Justification>)>, finalized_blocks: Vec<(BlockId<Block>, Option<Justification>)>,
set_head: Option<BlockId<Block>>, set_head: Option<BlockId<Block>>,
commit_state: bool,
} }
impl<Block: BlockT, H: Hasher> BlockImportOperation<Block, H> { impl<Block: BlockT, H: Hasher> BlockImportOperation<Block, H> {
@@ -531,6 +532,7 @@ impl<Block> client::backend::BlockImportOperation<Block, Blake2Hasher>
); );
self.db_updates = transaction; self.db_updates = transaction;
self.commit_state = true;
Ok(root) Ok(root)
} }
@@ -783,6 +785,7 @@ pub struct Backend<Block: BlockT> {
canonicalization_delay: u64, canonicalization_delay: u64,
shared_cache: SharedCache<Block, Blake2Hasher>, shared_cache: SharedCache<Block, Blake2Hasher>,
import_lock: Mutex<()>, import_lock: Mutex<()>,
is_archive: bool,
} }
impl<Block: BlockT<Hash=H256>> Backend<Block> { impl<Block: BlockT<Hash=H256>> Backend<Block> {
@@ -843,6 +846,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
config.state_cache_child_ratio.unwrap_or(DEFAULT_CHILD_RATIO), config.state_cache_child_ratio.unwrap_or(DEFAULT_CHILD_RATIO),
), ),
import_lock: Default::default(), import_lock: Default::default(),
is_archive: is_archive_pruning,
}) })
} }
@@ -894,6 +898,12 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
inmem inmem
} }
/// Returns total numbet of blocks (headers) in the block DB.
#[cfg(feature = "test-helpers")]
pub fn blocks_count(&self) -> u64 {
self.blockchain.db.iter(columns::HEADER).count() as u64
}
/// Read (from storage or cache) changes trie config. /// Read (from storage or cache) changes trie config.
/// ///
/// Currently changes tries configuration is set up once (at genesis) and could not /// Currently changes tries configuration is set up once (at genesis) and could not
@@ -1115,7 +1125,7 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
); );
transaction.put(columns::HEADER, &lookup_key, &pending_block.header.encode()); transaction.put(columns::HEADER, &lookup_key, &pending_block.header.encode());
if let Some(body) = pending_block.body { if let Some(body) = &pending_block.body {
transaction.put(columns::BODY, &lookup_key, &body.encode()); transaction.put(columns::BODY, &lookup_key, &body.encode());
} }
if let Some(justification) = pending_block.justification { if let Some(justification) = pending_block.justification {
@@ -1127,21 +1137,26 @@ impl<Block: BlockT<Hash=H256>> Backend<Block> {
transaction.put(columns::META, meta_keys::GENESIS_HASH, hash.as_ref()); transaction.put(columns::META, meta_keys::GENESIS_HASH, hash.as_ref());
} }
let mut changeset: state_db::ChangeSet<Vec<u8>> = state_db::ChangeSet::default(); let finalized = if operation.commit_state {
for (key, (val, rc)) in operation.db_updates.drain() { let mut changeset: state_db::ChangeSet<Vec<u8>> = state_db::ChangeSet::default();
if rc > 0 { for (key, (val, rc)) in operation.db_updates.drain() {
changeset.inserted.push((key, val.to_vec())); if rc > 0 {
} else if rc < 0 { changeset.inserted.push((key, val.to_vec()));
changeset.deleted.push(key); } else if rc < 0 {
changeset.deleted.push(key);
}
} }
} let number_u64 = number.saturated_into::<u64>();
let number_u64 = number.saturated_into::<u64>(); let commit = self.storage.state_db.insert_block(&hash, number_u64, &pending_block.header.parent_hash(), changeset)
let commit = self.storage.state_db.insert_block(&hash, number_u64, &pending_block.header.parent_hash(), changeset) .map_err(|e: state_db::Error<io::Error>| client::error::Error::from(format!("State database error: {:?}", e)))?;
.map_err(|e: state_db::Error<io::Error>| client::error::Error::from(format!("State database error: {:?}", e)))?; apply_state_commit(&mut transaction, commit);
apply_state_commit(&mut transaction, commit);
// Check if need to finalize. Genesis is always finalized instantly. // Check if need to finalize. Genesis is always finalized instantly.
let finalized = number_u64 == 0 || pending_block.leaf_state.is_final(); let finalized = number_u64 == 0 || pending_block.leaf_state.is_final();
finalized
} else {
false
};
let header = &pending_block.header; let header = &pending_block.header;
let is_best = pending_block.leaf_state.is_best(); let is_best = pending_block.leaf_state.is_best();
@@ -1347,6 +1362,7 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
aux_ops: Vec::new(), aux_ops: Vec::new(),
finalized_blocks: Vec::new(), finalized_blocks: Vec::new(),
set_head: None, set_head: None,
commit_state: false,
}) })
} }
@@ -1356,6 +1372,7 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
block: BlockId<Block>, block: BlockId<Block>,
) -> ClientResult<()> { ) -> ClientResult<()> {
operation.old_state = self.state_at(block)?; operation.old_state = self.state_at(block)?;
operation.commit_state = true;
Ok(()) Ok(())
} }
@@ -1478,6 +1495,9 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
match self.blockchain.header(block) { match self.blockchain.header(block) {
Ok(Some(ref hdr)) => { Ok(Some(ref hdr)) => {
let hash = hdr.hash(); let hash = hdr.hash();
if !self.have_state_at(&hash, *hdr.number()) {
return Err(client::error::Error::UnknownBlock(format!("State already discarded for {:?}", block)))
}
if let Ok(()) = self.storage.state_db.pin(&hash) { if let Ok(()) = self.storage.state_db.pin(&hash) {
let root = H256::from_slice(hdr.state_root().as_ref()); let root = H256::from_slice(hdr.state_root().as_ref());
let db_state = DbState::new(self.storage.clone(), root); let db_state = DbState::new(self.storage.clone(), root);
@@ -1493,7 +1513,16 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
} }
fn have_state_at(&self, hash: &Block::Hash, number: NumberFor<Block>) -> bool { fn have_state_at(&self, hash: &Block::Hash, number: NumberFor<Block>) -> bool {
!self.storage.state_db.is_pruned(hash, number.saturated_into::<u64>()) if self.is_archive {
match self.blockchain.header(BlockId::Hash(hash.clone())) {
Ok(Some(header)) => {
state_machine::Storage::get(self.storage.as_ref(), &header.state_root(), (&[], None)).unwrap_or(None).is_some()
},
_ => false,
}
} else {
!self.storage.state_db.is_pruned(hash, number.saturated_into::<u64>())
}
} }
fn destroy_state(&self, state: Self::State) -> ClientResult<()> { fn destroy_state(&self, state: Self::State) -> ClientResult<()> {
@@ -1581,7 +1610,7 @@ mod tests {
}; };
let mut op = backend.begin_operation().unwrap(); let mut op = backend.begin_operation().unwrap();
backend.begin_state_operation(&mut op, block_id).unwrap(); backend.begin_state_operation(&mut op, block_id).unwrap();
op.set_block_data(header, None, None, NewBlockState::Best).unwrap(); op.set_block_data(header, Some(Vec::new()), None, NewBlockState::Best).unwrap();
op.update_changes_trie((changes_trie_update, ChangesTrieCacheAction::Clear)).unwrap(); op.update_changes_trie((changes_trie_update, ChangesTrieCacheAction::Clear)).unwrap();
backend.commit_operation(op).unwrap(); backend.commit_operation(op).unwrap();
+49 -33
View File
@@ -838,15 +838,22 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
finalized, finalized,
auxiliary, auxiliary,
fork_choice, fork_choice,
allow_missing_state,
} = import_block; } = import_block;
assert!(justification.is_some() && finalized || justification.is_none()); assert!(justification.is_some() && finalized || justification.is_none());
let parent_hash = header.parent_hash().clone(); let parent_hash = header.parent_hash().clone();
let mut enact_state = true;
match self.backend.blockchain().status(BlockId::Hash(parent_hash))? { match self.block_status(&BlockId::Hash(parent_hash))? {
blockchain::BlockStatus::InChain => {}, BlockStatus::Unknown => return Ok(ImportResult::UnknownParent),
blockchain::BlockStatus::Unknown => return Ok(ImportResult::UnknownParent), BlockStatus::InChainWithState | BlockStatus::Queued => {},
BlockStatus::InChainPruned if allow_missing_state => {
enact_state = false;
},
BlockStatus::InChainPruned => return Ok(ImportResult::MissingState),
BlockStatus::KnownBad => return Ok(ImportResult::KnownBad),
} }
let import_headers = if post_digests.is_empty() { let import_headers = if post_digests.is_empty() {
@@ -875,6 +882,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
finalized, finalized,
auxiliary, auxiliary,
fork_choice, fork_choice,
enact_state,
); );
if let Ok(ImportResult::Imported(ref aux)) = result { if let Ok(ImportResult::Imported(ref aux)) = result {
@@ -902,6 +910,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
finalized: bool, finalized: bool,
aux: Vec<(Vec<u8>, Option<Vec<u8>>)>, aux: Vec<(Vec<u8>, Option<Vec<u8>>)>,
fork_choice: ForkChoiceStrategy, fork_choice: ForkChoiceStrategy,
enact_state: bool,
) -> error::Result<ImportResult> where ) -> error::Result<ImportResult> where
E: CallExecutor<Block, Blake2Hasher> + Send + Sync + Clone, E: CallExecutor<Block, Blake2Hasher> + Send + Sync + Clone,
{ {
@@ -927,22 +936,39 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
BlockOrigin::Genesis | BlockOrigin::NetworkInitialSync | BlockOrigin::File => false, BlockOrigin::Genesis | BlockOrigin::NetworkInitialSync | BlockOrigin::File => false,
}; };
self.backend.begin_state_operation(&mut operation.op, BlockId::Hash(parent_hash))?; let storage_changes = match &body {
Some(body) if enact_state => {
self.backend.begin_state_operation(&mut operation.op, BlockId::Hash(parent_hash))?;
// ensure parent block is finalized to maintain invariant that // ensure parent block is finalized to maintain invariant that
// finality is called sequentially. // finality is called sequentially.
if finalized { if finalized {
self.apply_finality_with_block_hash(operation, parent_hash, None, info.best_hash, make_notifications)?; self.apply_finality_with_block_hash(operation, parent_hash, None, info.best_hash, make_notifications)?;
} }
// FIXME #1232: correct path logic for when to execute this function // FIXME #1232: correct path logic for when to execute this function
let (storage_update, changes_update, storage_changes) = self.block_execution( let (storage_update, changes_update, storage_changes) = self.block_execution(
&operation.op, &operation.op,
&import_headers, &import_headers,
origin, origin,
hash, hash,
body.clone(), &body,
)?; )?;
operation.op.update_cache(new_cache);
if let Some(storage_update) = storage_update {
operation.op.update_db_storage(storage_update)?;
}
if let Some(storage_changes) = storage_changes.clone() {
operation.op.update_storage(storage_changes.0, storage_changes.1)?;
}
if let Some(Some(changes_update)) = changes_update {
operation.op.update_changes_trie(changes_update)?;
}
storage_changes
},
_ => None,
};
let is_new_best = finalized || match fork_choice { let is_new_best = finalized || match fork_choice {
ForkChoiceStrategy::LongestChain => import_headers.post().number() > &info.best_number, ForkChoiceStrategy::LongestChain => import_headers.post().number() > &info.best_number,
@@ -977,17 +1003,6 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
leaf_state, leaf_state,
)?; )?;
operation.op.update_cache(new_cache);
if let Some(storage_update) = storage_update {
operation.op.update_db_storage(storage_update)?;
}
if let Some(storage_changes) = storage_changes.clone() {
operation.op.update_storage(storage_changes.0, storage_changes.1)?;
}
if let Some(Some(changes_update)) = changes_update {
operation.op.update_changes_trie(changes_update)?;
}
operation.op.insert_aux(aux)?; operation.op.insert_aux(aux)?;
if make_notifications { if make_notifications {
@@ -1014,7 +1029,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
import_headers: &PrePostHeader<Block::Header>, import_headers: &PrePostHeader<Block::Header>,
origin: BlockOrigin, origin: BlockOrigin,
hash: Block::Hash, hash: Block::Hash,
body: Option<Vec<Block::Extrinsic>>, body: &[Block::Extrinsic],
) -> error::Result<( ) -> error::Result<(
Option<StorageUpdate<B, Block>>, Option<StorageUpdate<B, Block>>,
Option<Option<ChangesUpdate<Block>>>, Option<Option<ChangesUpdate<Block>>>,
@@ -1052,7 +1067,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
let encoded_block = <Block as BlockT>::encode_from( let encoded_block = <Block as BlockT>::encode_from(
import_headers.pre(), import_headers.pre(),
&body.unwrap_or_default() body,
); );
let (_, storage_update, changes_update) = self.executor let (_, storage_update, changes_update) = self.executor
@@ -1523,7 +1538,7 @@ impl<'a, B, E, Block, RA> consensus::BlockImport<Block> for &'a Client<B, E, Blo
&mut self, &mut self,
block: BlockCheckParams<Block>, block: BlockCheckParams<Block>,
) -> Result<ImportResult, Self::Error> { ) -> Result<ImportResult, Self::Error> {
let BlockCheckParams { hash, number, parent_hash } = block; let BlockCheckParams { hash, number, parent_hash, allow_missing_state } = block;
if let Some(h) = self.fork_blocks.as_ref().and_then(|x| x.get(&number)) { if let Some(h) = self.fork_blocks.as_ref().and_then(|x| x.get(&number)) {
if &hash != h { if &hash != h {
@@ -1541,7 +1556,9 @@ impl<'a, B, E, Block, RA> consensus::BlockImport<Block> for &'a Client<B, E, Blo
.map_err(|e| ConsensusError::ClientImport(e.to_string()))? .map_err(|e| ConsensusError::ClientImport(e.to_string()))?
{ {
BlockStatus::InChainWithState | BlockStatus::Queued => {}, BlockStatus::InChainWithState | BlockStatus::Queued => {},
BlockStatus::Unknown | BlockStatus::InChainPruned => return Ok(ImportResult::UnknownParent), BlockStatus::Unknown => return Ok(ImportResult::UnknownParent),
BlockStatus::InChainPruned if allow_missing_state => {},
BlockStatus::InChainPruned => return Ok(ImportResult::MissingState),
BlockStatus::KnownBad => return Ok(ImportResult::KnownBad), BlockStatus::KnownBad => return Ok(ImportResult::KnownBad),
} }
@@ -1553,7 +1570,6 @@ impl<'a, B, E, Block, RA> consensus::BlockImport<Block> for &'a Client<B, E, Blo
BlockStatus::KnownBad => return Ok(ImportResult::KnownBad), BlockStatus::KnownBad => return Ok(ImportResult::KnownBad),
} }
Ok(ImportResult::imported(false)) Ok(ImportResult::imported(false))
} }
} }
+2
View File
@@ -267,6 +267,7 @@ impl<H, B, C, E, I, P, Error, SO> slots::SimpleSlotWorker<B> for AuraWorker<C, E
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
} }
}) })
} }
@@ -570,6 +571,7 @@ impl<B: BlockT, C, P, T> Verifier<B> for AuraVerifier<C, P, T> where
justification, justification,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
Ok((block_import_params, maybe_keys)) Ok((block_import_params, maybe_keys))
+2
View File
@@ -430,6 +430,7 @@ impl<B, C, E, I, Error, SO> slots::SimpleSlotWorker<B> for BabeWorker<B, C, E, I
// option to specify one. // option to specify one.
// https://github.com/paritytech/substrate/issues/3623 // https://github.com/paritytech/substrate/issues/3623
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
} }
}) })
} }
@@ -741,6 +742,7 @@ impl<B, E, Block, RA, PRA> Verifier<Block> for BabeVerifier<B, E, Block, RA, PRA
// option to specify one. // option to specify one.
// https://github.com/paritytech/substrate/issues/3623 // https://github.com/paritytech/substrate/issues/3623
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
Ok((block_import_params, Default::default())) Ok((block_import_params, Default::default()))
@@ -578,6 +578,7 @@ fn propose_and_import_block(
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}, },
Default::default(), Default::default(),
).unwrap(); ).unwrap();
@@ -35,11 +35,15 @@ pub enum ImportResult {
KnownBad, KnownBad,
/// Block parent is not in the chain. /// Block parent is not in the chain.
UnknownParent, UnknownParent,
/// Parent state is missing.
MissingState,
} }
/// Auxiliary data associated with an imported block result. /// Auxiliary data associated with an imported block result.
#[derive(Debug, Default, PartialEq, Eq)] #[derive(Debug, Default, PartialEq, Eq)]
pub struct ImportedAux { pub struct ImportedAux {
/// Only the header has been imported. Block body verification was skipped.
pub header_only: bool,
/// Clear all pending justification requests. /// Clear all pending justification requests.
pub clear_justification_requests: bool, pub clear_justification_requests: bool,
/// Request a justification for the given block. /// Request a justification for the given block.
@@ -98,6 +102,8 @@ pub struct BlockCheckParams<Block: BlockT> {
pub number: NumberFor<Block>, pub number: NumberFor<Block>,
/// Parent hash of the block that we verify. /// Parent hash of the block that we verify.
pub parent_hash: Block::Hash, pub parent_hash: Block::Hash,
/// Allow importing the block skipping state verification if parent state is missing.
pub allow_missing_state: bool,
} }
/// Data required to import a Block. /// Data required to import a Block.
@@ -133,6 +139,8 @@ pub struct BlockImportParams<Block: BlockT> {
/// Fork choice strategy of this import. This should only be set by a /// Fork choice strategy of this import. This should only be set by a
/// synchronous import, otherwise it may race against other imports. /// synchronous import, otherwise it may race against other imports.
pub fork_choice: ForkChoiceStrategy, pub fork_choice: ForkChoiceStrategy,
/// Allow importing the block skipping state verification if parent state is missing.
pub allow_missing_state: bool,
} }
impl<Block: BlockT> BlockImportParams<Block> { impl<Block: BlockT> BlockImportParams<Block> {
@@ -63,6 +63,8 @@ pub struct IncomingBlock<B: BlockT> {
pub justification: Option<Justification>, pub justification: Option<Justification>,
/// The peer, we received this from /// The peer, we received this from
pub origin: Option<Origin>, pub origin: Option<Origin>,
/// Allow importing the block skipping state verification if parent state is missing.
pub allow_missing_state: bool,
} }
/// Type of keys in the blockchain cache that consensus module could use for its needs. /// Type of keys in the blockchain cache that consensus module could use for its needs.
@@ -203,6 +205,10 @@ pub fn import_single_block<B: BlockT, V: Verifier<B>>(
Ok(BlockImportResult::ImportedKnown(number)) Ok(BlockImportResult::ImportedKnown(number))
}, },
Ok(ImportResult::Imported(aux)) => Ok(BlockImportResult::ImportedUnknown(number, aux, peer.clone())), Ok(ImportResult::Imported(aux)) => Ok(BlockImportResult::ImportedUnknown(number, aux, peer.clone())),
Ok(ImportResult::MissingState) => {
debug!(target: "sync", "Parent state is missing for {}: {:?}, parent: {:?}", number, hash, parent_hash);
Err(BlockImportError::UnknownParent)
},
Ok(ImportResult::UnknownParent) => { Ok(ImportResult::UnknownParent) => {
debug!(target: "sync", "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent_hash); debug!(target: "sync", "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent_hash);
Err(BlockImportError::UnknownParent) Err(BlockImportError::UnknownParent)
@@ -217,12 +223,17 @@ pub fn import_single_block<B: BlockT, V: Verifier<B>>(
} }
} }
}; };
match import_error(import_handle.check_block(BlockCheckParams { hash, number, parent_hash }))? { match import_error(import_handle.check_block(BlockCheckParams {
hash,
number,
parent_hash,
allow_missing_state: block.allow_missing_state,
}))? {
BlockImportResult::ImportedUnknown { .. } => (), BlockImportResult::ImportedUnknown { .. } => (),
r => return Ok(r), // Any other successful result means that the block is already imported. r => return Ok(r), // Any other successful result means that the block is already imported.
} }
let (import_block, maybe_keys) = verifier.verify(block_origin, header, justification, block.body) let (mut import_block, maybe_keys) = verifier.verify(block_origin, header, justification, block.body)
.map_err(|msg| { .map_err(|msg| {
if let Some(ref peer) = peer { if let Some(ref peer) = peer {
trace!(target: "sync", "Verifying {}({}) from {} failed: {}", number, hash, peer, msg); trace!(target: "sync", "Verifying {}({}) from {} failed: {}", number, hash, peer, msg);
@@ -236,6 +247,7 @@ pub fn import_single_block<B: BlockT, V: Verifier<B>>(
if let Some(keys) = maybe_keys { if let Some(keys) = maybe_keys {
cache.extend(keys.into_iter()); cache.extend(keys.into_iter());
} }
import_block.allow_missing_state = block.allow_missing_state;
import_error(import_handle.import_block(import_block, cache)) import_error(import_handle.import_block(import_block, cache))
} }
+2
View File
@@ -304,6 +304,7 @@ impl<B: BlockT<Hash=H256>, C, S, Algorithm> Verifier<B> for PowVerifier<B, C, S,
justification, justification,
auxiliary: vec![(key, Some(aux.encode()))], auxiliary: vec![(key, Some(aux.encode()))],
fork_choice: ForkChoiceStrategy::Custom(aux.total_difficulty > best_aux.total_difficulty), fork_choice: ForkChoiceStrategy::Custom(aux.total_difficulty > best_aux.total_difficulty),
allow_missing_state: false,
}; };
Ok((import_block, None)) Ok((import_block, None))
@@ -531,6 +532,7 @@ fn mine_loop<B: BlockT<Hash=H256>, C, Algorithm, E, SO, S>(
finalized: false, finalized: false,
auxiliary: vec![(key, Some(aux.encode()))], auxiliary: vec![(key, Some(aux.encode()))],
fork_choice: ForkChoiceStrategy::Custom(true), fork_choice: ForkChoiceStrategy::Custom(true),
allow_missing_state: false,
}; };
block_import.import_block(import_block, HashMap::default()) block_import.import_block(import_block, HashMap::default())
@@ -663,6 +663,7 @@ pub mod tests {
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: true,
}; };
do_import_block::<_, _, _, TestJustification>( do_import_block::<_, _, _, TestJustification>(
&client, &client,
@@ -680,6 +681,7 @@ pub mod tests {
bad_justification: false, bad_justification: false,
needs_finality_proof: false, needs_finality_proof: false,
is_new_best: true, is_new_best: true,
header_only: false,
})); }));
} }
@@ -692,6 +694,7 @@ pub mod tests {
bad_justification: false, bad_justification: false,
needs_finality_proof: false, needs_finality_proof: false,
is_new_best: true, is_new_best: true,
header_only: false,
})); }));
} }
@@ -705,6 +708,7 @@ pub mod tests {
bad_justification: false, bad_justification: false,
needs_finality_proof: true, needs_finality_proof: true,
is_new_best: true, is_new_best: true,
header_only: false,
})); }));
} }
@@ -721,6 +725,7 @@ pub mod tests {
bad_justification: false, bad_justification: false,
needs_finality_proof: true, needs_finality_proof: true,
is_new_best: false, is_new_best: false,
header_only: false,
}, },
)); ));
} }
@@ -974,6 +974,7 @@ fn allows_reimporting_change_blocks() {
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
} }
}; };
@@ -985,6 +986,7 @@ fn allows_reimporting_change_blocks() {
bad_justification: false, bad_justification: false,
needs_finality_proof: false, needs_finality_proof: false,
is_new_best: true, is_new_best: true,
header_only: false,
}), }),
); );
@@ -1025,6 +1027,7 @@ fn test_bad_justification() {
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
} }
}; };
@@ -1721,6 +1724,7 @@ fn imports_justification_for_regular_blocks_on_import() {
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
assert_eq!( assert_eq!(
+27 -18
View File
@@ -480,13 +480,7 @@ impl<B: BlockT> ChainSync<B> {
return; return;
} }
let block_status = self.client.block_status(&BlockId::Number(number - One::one())) trace!(target: "sync", "Downloading requested old fork {:?}", hash);
.unwrap_or(BlockStatus::Unknown);
if block_status == BlockStatus::InChainPruned {
trace!(target: "sync", "Refusing to sync ancient block {:?}", hash);
return;
}
self.is_idle = false; self.is_idle = false;
for peer_id in &peers { for peer_id in &peers {
if let Some(peer) = self.peers.get_mut(peer_id) { if let Some(peer) = self.peers.get_mut(peer_id) {
@@ -571,7 +565,7 @@ impl<B: BlockT> ChainSync<B> {
let major_sync = self.status().state == SyncState::Downloading; let major_sync = self.status().state == SyncState::Downloading;
let blocks = &mut self.blocks; let blocks = &mut self.blocks;
let attrs = &self.required_block_attributes; let attrs = &self.required_block_attributes;
let fork_targets = &self.fork_targets; let fork_targets = &mut self.fork_targets;
let mut have_requests = false; let mut have_requests = false;
let last_finalized = self.client.info().chain.finalized_number; let last_finalized = self.client.info().chain.finalized_number;
let best_queued = self.best_queued_number; let best_queued = self.best_queued_number;
@@ -646,6 +640,7 @@ impl<B: BlockT> ChainSync<B> {
body: block_data.block.body, body: block_data.block.body,
justification: block_data.block.justification, justification: block_data.block.justification,
origin: block_data.origin, origin: block_data.origin,
allow_missing_state: false,
} }
}).collect() }).collect()
} }
@@ -658,14 +653,15 @@ impl<B: BlockT> ChainSync<B> {
body: b.body, body: b.body,
justification: b.justification, justification: b.justification,
origin: Some(who.clone()), origin: Some(who.clone()),
allow_missing_state: true,
} }
}).collect() }).collect()
} }
PeerSyncState::AncestorSearch(num, state) => { PeerSyncState::AncestorSearch(num, state) => {
let block_hash_match = match (blocks.get(0), self.client.block_hash(*num)) { let matching_hash = match (blocks.get(0), self.client.block_hash(*num)) {
(Some(block), Ok(maybe_our_block_hash)) => { (Some(block), Ok(maybe_our_block_hash)) => {
trace!(target: "sync", "Got ancestry block #{} ({}) from peer {}", num, block.hash, who); trace!(target: "sync", "Got ancestry block #{} ({}) from peer {}", num, block.hash, who);
maybe_our_block_hash.map_or(false, |x| x == block.hash) maybe_our_block_hash.filter(|x| x == &block.hash)
}, },
(None, _) => { (None, _) => {
debug!(target: "sync", "Invalid response when searching for ancestor from {}", who); debug!(target: "sync", "Invalid response when searching for ancestor from {}", who);
@@ -676,27 +672,34 @@ impl<B: BlockT> ChainSync<B> {
return Err(BadPeer(who, ANCESTRY_BLOCK_ERROR_REPUTATION_CHANGE)) return Err(BadPeer(who, ANCESTRY_BLOCK_ERROR_REPUTATION_CHANGE))
} }
}; };
if block_hash_match && peer.common_number < *num { if matching_hash.is_some() && peer.common_number < *num {
peer.common_number = *num; peer.common_number = *num;
} }
if !block_hash_match && num.is_zero() { if matching_hash.is_none() && num.is_zero() {
trace!(target:"sync", "Ancestry search: genesis mismatch for peer {}", who); trace!(target:"sync", "Ancestry search: genesis mismatch for peer {}", who);
return Err(BadPeer(who, GENESIS_MISMATCH_REPUTATION_CHANGE)) return Err(BadPeer(who, GENESIS_MISMATCH_REPUTATION_CHANGE))
} }
if let Some((next_state, next_num)) = handle_ancestor_search_state(state, *num, block_hash_match) { if let Some((next_state, next_num)) = handle_ancestor_search_state(state, *num, matching_hash.is_some()) {
peer.state = PeerSyncState::AncestorSearch(next_num, next_state); peer.state = PeerSyncState::AncestorSearch(next_num, next_state);
return Ok(OnBlockData::Request(who, ancestry_request::<B>(next_num))) return Ok(OnBlockData::Request(who, ancestry_request::<B>(next_num)))
} else { } else {
// Ancestry search is complete. Check if peer is on a stale fork unknown to us and // Ancestry search is complete. Check if peer is on a stale fork unknown to us and
// add it to sync targets if necessary. // add it to sync targets if necessary.
trace!(target: "sync", "Ancestry search complete. Ours={} ({}), Theirs={} ({}), Common={}", trace!(target: "sync", "Ancestry search complete. Ours={} ({}), Theirs={} ({}), Common={:?} ({})",
self.best_queued_hash, self.best_queued_hash,
self.best_queued_number, self.best_queued_number,
peer.best_hash, peer.best_hash,
peer.best_number, peer.best_number,
peer.common_number matching_hash,
peer.common_number,
); );
if peer.common_number < peer.best_number && peer.best_number < self.best_queued_number { let client = &self.client;
if peer.common_number < peer.best_number
&& peer.best_number < self.best_queued_number
&& matching_hash.and_then(
|h| client.block_status(&BlockId::Hash(h)).ok()
).unwrap_or(BlockStatus::Unknown) != BlockStatus::InChainPruned
{
trace!(target: "sync", "Added fork target {} for {}" , peer.best_hash, who); trace!(target: "sync", "Added fork target {} for {}" , peer.best_hash, who);
self.fork_targets self.fork_targets
.entry(peer.best_hash.clone()) .entry(peer.best_hash.clone())
@@ -1250,25 +1253,31 @@ fn peer_block_request<B: BlockT>(
/// Get pending fork sync targets for a peer. /// Get pending fork sync targets for a peer.
fn fork_sync_request<B: BlockT>( fn fork_sync_request<B: BlockT>(
id: &PeerId, id: &PeerId,
targets: &HashMap<B::Hash, ForkTarget<B>>, targets: &mut HashMap<B::Hash, ForkTarget<B>>,
best_num: NumberFor<B>, best_num: NumberFor<B>,
finalized: NumberFor<B>, finalized: NumberFor<B>,
attributes: &message::BlockAttributes, attributes: &message::BlockAttributes,
check_block: impl Fn(&B::Hash) -> BlockStatus, check_block: impl Fn(&B::Hash) -> BlockStatus,
) -> Option<(B::Hash, BlockRequest<B>)> ) -> Option<(B::Hash, BlockRequest<B>)>
{ {
targets.retain(|hash, r| if r.number > finalized {
true
} else {
trace!(target: "sync", "Removed expired fork sync request {:?} (#{})", hash, r.number);
false
});
for (hash, r) in targets { for (hash, r) in targets {
if !r.peers.contains(id) { if !r.peers.contains(id) {
continue continue
} }
if r.number <= best_num { if r.number <= best_num {
trace!(target: "sync", "Downloading requested fork {:?} from {}", hash, id);
let parent_status = r.parent_hash.as_ref().map_or(BlockStatus::Unknown, check_block); let parent_status = r.parent_hash.as_ref().map_or(BlockStatus::Unknown, check_block);
let mut count = (r.number - finalized).saturated_into::<u32>(); // up to the last finalized block let mut count = (r.number - finalized).saturated_into::<u32>(); // up to the last finalized block
if parent_status != BlockStatus::Unknown { if parent_status != BlockStatus::Unknown {
// request only single block // request only single block
count = 1; count = 1;
} }
trace!(target: "sync", "Downloading requested fork {:?} from {}, {} blocks", hash, id, count);
return Some((hash.clone(), message::generic::BlockRequest { return Some((hash.clone(), message::generic::BlockRequest {
id: 0, id: 0,
fields: attributes.clone(), fields: attributes.clone(),
@@ -37,9 +37,10 @@ fn prepare_good_block() -> (TestClient, Hash, u64, PeerId, IncomingBlock<Block>)
(client, hash, number, peer_id.clone(), IncomingBlock { (client, hash, number, peer_id.clone(), IncomingBlock {
hash, hash,
header, header,
body: None, body: Some(Vec::new()),
justification, justification,
origin: Some(peer_id.clone()) origin: Some(peer_id.clone()),
allow_missing_state: false,
}) })
} }
@@ -53,7 +54,7 @@ fn import_single_good_block_works() {
match import_single_block(&mut test_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier(true)) { match import_single_block(&mut test_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier(true)) {
Ok(BlockImportResult::ImportedUnknown(ref num, ref aux, ref org)) Ok(BlockImportResult::ImportedUnknown(ref num, ref aux, ref org))
if *num == number && *aux == expected_aux && *org == Some(peer_id) => {} if *num == number && *aux == expected_aux && *org == Some(peer_id) => {}
_ => panic!() r @ _ => panic!("{:?}", r)
} }
} }
+14 -12
View File
@@ -96,6 +96,7 @@ impl<B: BlockT> Verifier<B> for PassThroughVerifier {
post_digests: vec![], post_digests: vec![],
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}, maybe_keys)) }, maybe_keys))
} }
} }
@@ -374,16 +375,10 @@ impl<D, S: NetworkSpecialization<Block>> Peer<D, S> {
} }
} }
/// Count the current number of known blocks. Note that: /// Count the total number of imported blocks.
/// 1. this might be expensive as it creates an in-memory-copy of the chain pub fn blocks_count(&self) -> u64 {
/// to count the blocks, thus if you have a different way of testing this
/// (e.g. `info.best_hash`) - use that.
/// 2. This is not always increasing nor accurate, as the
/// orphaned and proven-to-never-finalized blocks may be pruned at any time.
/// Therefore, this number can drop again.
pub fn blocks_count(&self) -> usize {
self.backend.as_ref().map( self.backend.as_ref().map(
|backend| backend.as_in_memory().blockchain().blocks_count() |backend| backend.blocks_count()
).unwrap_or(0) ).unwrap_or(0)
} }
} }
@@ -526,9 +521,16 @@ pub trait TestNetFactory: Sized {
net net
} }
/// Add a full peer.
fn add_full_peer(&mut self, config: &ProtocolConfig) { fn add_full_peer(&mut self, config: &ProtocolConfig) {
let test_client_builder = TestClientBuilder::with_default_backend(); self.add_full_peer_with_states(config, None)
}
/// Add a full peer.
fn add_full_peer_with_states(&mut self, config: &ProtocolConfig, keep_blocks: Option<u32>) {
let test_client_builder = match keep_blocks {
Some(keep_blocks) => TestClientBuilder::with_pruning_window(keep_blocks),
None => TestClientBuilder::with_default_backend(),
};
let backend = test_client_builder.backend(); let backend = test_client_builder.backend();
let (c, longest_chain) = test_client_builder.build_with_longest_chain(); let (c, longest_chain) = test_client_builder.build_with_longest_chain();
let client = Arc::new(c); let client = Arc::new(c);
@@ -686,7 +688,7 @@ pub trait TestNetFactory: Sized {
if peer.is_major_syncing() || peer.network.num_queued_blocks() != 0 { if peer.is_major_syncing() || peer.network.num_queued_blocks() != 0 {
return Async::NotReady return Async::NotReady
} }
match (highest, peer.client.info().chain.best_number) { match (highest, peer.client.info().chain.best_hash) {
(None, b) => highest = Some(b), (None, b) => highest = Some(b),
(Some(ref a), ref b) if a == b => {}, (Some(ref a), ref b) if a == b => {},
(Some(_), _) => return Async::NotReady, (Some(_), _) => return Async::NotReady,
+42 -1
View File
@@ -236,7 +236,14 @@ fn sync_no_common_longer_chain_fails() {
let mut net = TestNet::new(3); let mut net = TestNet::new(3);
net.peer(0).push_blocks(20, true); net.peer(0).push_blocks(20, true);
net.peer(1).push_blocks(20, false); net.peer(1).push_blocks(20, false);
net.block_until_sync(&mut runtime); runtime.block_on(futures::future::poll_fn::<(), (), _>(|| -> Result<_, ()> {
net.poll();
if net.peer(0).is_major_syncing() {
Ok(Async::NotReady)
} else {
Ok(Async::Ready(()))
}
})).unwrap();
let peer1 = &net.peers()[1]; let peer1 = &net.peers()[1];
assert!(!net.peers()[0].blockchain_canon_equals(peer1)); assert!(!net.peers()[0].blockchain_canon_equals(peer1));
} }
@@ -592,3 +599,37 @@ fn can_sync_explicit_forks() {
Ok(Async::Ready(())) Ok(Async::Ready(()))
})).unwrap(); })).unwrap();
} }
#[test]
fn syncs_header_only_forks() {
let _ = ::env_logger::try_init();
let mut runtime = current_thread::Runtime::new().unwrap();
let mut net = TestNet::new(0);
let config = ProtocolConfig::default();
net.add_full_peer_with_states(&config, None);
net.add_full_peer_with_states(&config, Some(3));
net.peer(0).push_blocks(2, false);
net.peer(1).push_blocks(2, false);
net.peer(0).push_blocks(2, true);
let small_hash = net.peer(0).client().info().chain.best_hash;
let small_number = net.peer(0).client().info().chain.best_number;
net.peer(1).push_blocks(4, false);
net.block_until_sync(&mut runtime);
// Peer 1 won't sync the small fork because common block state is missing
assert_eq!(9, net.peer(0).blocks_count());
assert_eq!(7, net.peer(1).blocks_count());
// Request explicit header-only sync request for the ancient fork.
let first_peer_id = net.peer(0).id();
net.peer(1).set_sync_fork_request(vec![first_peer_id], small_hash, small_number);
runtime.block_on(futures::future::poll_fn::<(), (), _>(|| -> Result<_, ()> {
net.poll();
if net.peer(1).client().header(&BlockId::Hash(small_hash)).unwrap().is_none() {
return Ok(Async::NotReady)
}
Ok(Async::Ready(()))
})).unwrap();
}
+1
View File
@@ -156,6 +156,7 @@ macro_rules! import_blocks {
body: block.body, body: block.body,
justification: block.justification, justification: block.justification,
origin: None, origin: None,
allow_missing_state: false,
} }
]); ]);
} }
@@ -77,6 +77,7 @@ impl<B, E, RA, Block> ClientExt<Block> for Client<B, E, Block, RA>
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ()) BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ())
@@ -95,6 +96,7 @@ impl<B, E, RA, Block> ClientExt<Block> for Client<B, E, Block, RA>
finalized: false, finalized: false,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::Custom(true), fork_choice: ForkChoiceStrategy::Custom(true),
allow_missing_state: false,
}; };
BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ()) BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ())
@@ -116,6 +118,7 @@ impl<B, E, RA, Block> ClientExt<Block> for Client<B, E, Block, RA>
finalized: true, finalized: true,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ()) BlockImport::import_block(&mut (&*self), import, HashMap::new()).map(|_| ())
+6
View File
@@ -98,6 +98,12 @@ impl<Block, Executor, G: GenesisInit> TestClientBuilder<
pub fn backend(&self) -> Arc<Backend<Block>> { pub fn backend(&self) -> Arc<Backend<Block>> {
self.backend.clone() self.backend.clone()
} }
/// Create new `TestClientBuilder` with default backend and pruning window size
pub fn with_pruning_window(keep_blocks: u32) -> Self {
let backend = Arc::new(Backend::new_test(keep_blocks, 0));
Self::with_backend(backend)
}
} }
impl<Executor, Backend, G: GenesisInit> TestClientBuilder<Executor, Backend, G> { impl<Executor, Backend, G: GenesisInit> TestClientBuilder<Executor, Backend, G> {
-2
View File
@@ -1,5 +1,3 @@
use std::{env, path::PathBuf};
use vergen::{ConstantsFlags, generate_cargo_keys}; use vergen::{ConstantsFlags, generate_cargo_keys};
const ERROR_MSG: &str = "Failed to generate metadata files"; const ERROR_MSG: &str = "Failed to generate metadata files";
+1
View File
@@ -523,6 +523,7 @@ mod tests {
finalized: true, finalized: true,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
block_import.import_block(params, Default::default()) block_import.import_block(params, Default::default())
@@ -194,6 +194,7 @@ fn import_block<Backend, Exec, Block, RtApi>(
justification: None, justification: None,
auxiliary: Vec::new(), auxiliary: Vec::new(),
fork_choice: ForkChoiceStrategy::LongestChain, fork_choice: ForkChoiceStrategy::LongestChain,
allow_missing_state: false,
}; };
(&**client).import_block(import, HashMap::new()).expect("Failed to import block"); (&**client).import_block(import, HashMap::new()).expect("Failed to import block");
} }