Newer
Older
// Copyright 2017-2018 Parity Technologies (UK) Ltd.
// This file is part of Substrate.
// Substrate is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Substrate is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
//! Substrate Client
use parking_lot::{Mutex, RwLock};
use runtime_primitives::{
bft::Justification,
generic::{BlockId, SignedBlock, Block as RuntimeBlock},
transaction_validity::{TransactionValidity, TransactionTag},
};
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Zero, As, NumberFor, CurrentHeight, BlockNumberToHash};
use runtime_primitives::{ApplyResult, BuildStorage};
use runtime_api as api;
use primitives::{Blake2Hasher, H256, ChangesTrieConfiguration};
use primitives::storage::{StorageKey, StorageData};
use primitives::storage::well_known_keys;
use state_machine::{
ExecutionStrategy, ExecutionManager, prove_read,
key_changes, key_changes_proof, OverlayedChanges
use blockchain::{self, Info as ChainInfo, Backend as ChainBackend, HeaderBackend as ChainHeaderBackend};
use call_executor::{CallExecutor, LocalCallExecutor};
use notifications::{StorageNotifications, StorageEventStream};
/// Type that implements `futures::Stream` of block import events.
pub type ImportNotifications<Block> = mpsc::UnboundedReceiver<BlockImportNotification<Block>>;
/// A stream of block finality notifications.
pub type FinalityNotifications<Block> = mpsc::UnboundedReceiver<FinalityNotification<Block>>;
pub struct Client<B, E, Block> where Block: BlockT {
storage_notifications: Mutex<StorageNotifications<Block>>,
import_notification_sinks: Mutex<Vec<mpsc::UnboundedSender<BlockImportNotification<Block>>>>,
finality_notification_sinks: Mutex<Vec<mpsc::UnboundedSender<FinalityNotification<Block>>>>,
import_lock: Mutex<()>,
importing_block: RwLock<Option<Block::Hash>>, // holds the block hash currently being imported. TODO: replace this with block queue
block_execution_strategy: ExecutionStrategy,
api_execution_strategy: ExecutionStrategy,
changes_trie_config: Option<ChangesTrieConfiguration>,
/// Get block import event stream. Not guaranteed to be fired for every
/// imported block.
fn import_notification_stream(&self) -> ImportNotifications<Block>;
/// Get a stream of finality notifications. Not guaranteed to be fired for every
/// finalized block.
fn finality_notification_stream(&self) -> FinalityNotifications<Block>;
/// Get storage changes event stream.
///
/// Passing `None` as `filter_keys` subscribes to all storage changes.
fn storage_changes_notification_stream(&self, filter_keys: Option<&[StorageKey]>) -> error::Result<StorageEventStream<Block::Hash>>;
/// Chain head information.
/// Get best block header.
fn best_block_header(&self) -> Result<<Block as BlockT>::Header, error::Error>;
/// Fetch block body by ID.
pub trait BlockBody<Block: BlockT> {
/// Get block body by ID. Returns `None` if the body is not stored.
fn block_body(&self, id: &BlockId<Block>) -> error::Result<Option<Vec<<Block as BlockT>::Extrinsic>>>;
}
/// Client info
// TODO: split queue info from chain info and amalgamate into single struct.
#[derive(Debug)]
pub best_queued_number: Option<<<Block as BlockT>::Header as HeaderT>::Number>,
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
}
/// Block import result.
#[derive(Debug)]
pub enum ImportResult {
/// Added to the import queue.
Queued,
/// Already in the import queue.
AlreadyQueued,
/// Already in the blockchain.
AlreadyInChain,
/// Block or parent is known to be bad.
KnownBad,
/// Block parent is not in the chain.
UnknownParent,
}
/// Block status.
#[derive(Debug, PartialEq, Eq)]
pub enum BlockStatus {
/// Added to the import queue.
Queued,
/// Already in the blockchain.
InChain,
/// Block or parent is known to be bad.
KnownBad,
/// Not in the queue or the blockchain.
Unknown,
}
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum BlockOrigin {
/// Genesis block built into the client.
Genesis,
/// Block is part of the initial sync with the network.
NetworkInitialSync,
/// Block was broadcasted on the network.
NetworkBroadcast,
/// Block that was received from the network and validated in the consensus process.
ConsensusBroadcast,
/// Block that was collated by this node.
Own,
/// Block was imported from a file.
File,
}
/// Summary of an imported block
#[derive(Clone, Debug)]
pub struct BlockImportNotification<Block: BlockT> {
/// Imported block origin.
pub origin: BlockOrigin,
/// Imported block header.
/// Is this the new best block.
pub is_new_best: bool,
/// Tags provided by transactions imported in that block.
pub tags: Vec<TransactionTag>,
/// Summary of a finalized block.
#[derive(Clone, Debug)]
pub struct FinalityNotification<Block: BlockT> {
/// Imported block header hash.
pub hash: Block::Hash,
/// Imported block header.
pub header: Block::Header,
}
/// A header paired with a justification which has already been checked.
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct JustifiedHeader<Block: BlockT> {
header: <Block as BlockT>::Header,
justification: ::bft::Justification<Block::Hash>,
authorities: Vec<AuthorityId>,
/// Deconstruct the justified header into parts.
pub fn into_inner(self) -> (<Block as BlockT>::Header, ::bft::Justification<Block::Hash>, Vec<AuthorityId>) {
(self.header, self.justification, self.authorities)
}
}
/// Create an instance of in-memory client.
) -> error::Result<Client<in_mem::Backend<Block, Blake2Hasher>, LocalCallExecutor<in_mem::Backend<Block, Blake2Hasher>, E>, Block>>
E: CodeExecutor<Blake2Hasher> + RuntimeInfo,
Svyatoslav Nikolsky
committed
H256: From<Block::Hash>,
new_with_backend(Arc::new(in_mem::Backend::new()), executor, genesis_storage)
}
/// Create a client with the explicitely provided backend.
/// This is useful for testing backend implementations.
pub fn new_with_backend<B, E, Block, S>(
backend: Arc<B>,
executor: E,
build_genesis_storage: S,
) -> error::Result<Client<B, LocalCallExecutor<B, E>, Block>>
where
E: CodeExecutor<Blake2Hasher> + RuntimeInfo,
S: BuildStorage,
Block: BlockT,
H256: From<Block::Hash>,
B: backend::LocalBackend<Block, Blake2Hasher>
{
let call_executor = LocalCallExecutor::new(backend.clone(), executor);
Client::new(backend, call_executor, build_genesis_storage, ExecutionStrategy::NativeWhenPossible, ExecutionStrategy::NativeWhenPossible)
impl<B, E, Block> Client<B, E, Block> where
B: backend::Backend<Block, Blake2Hasher>,
E: CallExecutor<Block, Blake2Hasher>,
/// Creates new Substrate Client with given blockchain and code executor.
block_execution_strategy: ExecutionStrategy,
api_execution_strategy: ExecutionStrategy,
if backend.blockchain().header(BlockId::Number(Zero::zero()))?.is_none() {
let genesis_storage = build_genesis_storage.build_storage()?;
let genesis_block = genesis::construct_genesis_block::<Block>(&genesis_storage);
info!("Initialising Genesis block/state (state: {}, header-hash: {})", genesis_block.header().state_root(), genesis_block.header().hash());
let mut op = backend.begin_operation(BlockId::Hash(Default::default()))?;
op.reset_storage(genesis_storage.into_iter())?;
op.set_block_data(
genesis_block.deconstruct().0,
Some(vec![]),
None,
::backend::NewBlockState::Final
)?;
// changes trie configuration should never change => we can read it in advance
let changes_trie_config = backend.state_at(BlockId::Number(backend.blockchain().info()?.best_number))?
.storage(well_known_keys::CHANGES_TRIE_CONFIG)
.map_err(|e| error::Error::from_state(Box::new(e)))?
.and_then(|c| Decode::decode(&mut &*c));
storage_notifications: Default::default(),
import_notification_sinks: Default::default(),
finality_notification_sinks: Default::default(),
import_lock: Default::default(),
importing_block: Default::default(),
block_execution_strategy,
api_execution_strategy,
changes_trie_config,
})
}
/// Get a reference to the state at a given block.
pub fn state_at(&self, block: &BlockId<Block>) -> error::Result<B::State> {
self.backend.state_at(*block)
}
/// Expose backend reference. To be used in tests only
pub fn backend(&self) -> &Arc<B> {
&self.backend
}
/// Return single storage entry of contract under given address in state in a block of given hash.
pub fn storage(&self, id: &BlockId<Block>, key: &StorageKey) -> error::Result<Option<StorageData>> {
Ok(self.state_at(id)?
.storage(&key.0).map_err(|e| error::Error::from_state(Box::new(e)))?
.map(StorageData))
pub fn code_at(&self, id: &BlockId<Block>) -> error::Result<Vec<u8>> {
Ok(self.storage(id, &StorageKey(well_known_keys::CODE.to_vec()))?
.expect("None is returned if there's no value stored for the given key; ':code' key is always defined; qed").0)
/// Get the set of authorities at a given block.
pub fn authorities_at(&self, id: &BlockId<Block>) -> error::Result<Vec<AuthorityId>> {
match self.backend.blockchain().cache().and_then(|cache| cache.authorities_at(*id)) {
Some(cached_value) => Ok(cached_value),
None => self.executor.call(id, "authorities",&[])
.and_then(|r| Vec::<AuthorityId>::decode(&mut &r.return_data[..])
.ok_or(error::ErrorKind::InvalidAuthoritiesSet.into()))
/// Get the RuntimeVersion at a given block.
pub fn runtime_version_at(&self, id: &BlockId<Block>) -> error::Result<RuntimeVersion> {
// TODO: Post Poc-2 return an error if version is missing
self.executor.runtime_version(id)
/// Get call executor reference.
pub fn executor(&self) -> &E {
&self.executor
/// Returns the runtime metadata.
pub fn metadata(&self, id: &BlockId<Block>) -> error::Result<Vec<u8>> {
self.executor.call(id, "metadata",&[]).map(|v| v.return_data)
/// Reads storage value at a given block + key, returning read proof.
pub fn read_proof(&self, id: &BlockId<Block>, key: &[u8]) -> error::Result<Vec<Vec<u8>>> {
self.state_at(id)
.and_then(|state| prove_read(state, key)
.map(|(_, proof)| proof)
.map_err(Into::into))
}
/// Execute a call to a contract on top of state in a block of given hash
/// AND returning execution proof.
pub fn execution_proof(&self, id: &BlockId<Block>, method: &str, call_data: &[u8]) -> error::Result<(Vec<u8>, Vec<Vec<u8>>)> {
self.state_at(id).and_then(|state| self.executor.prove_at_state(state, &mut Default::default(), method, call_data))
/// Reads given header and generates CHT-based header proof.
pub fn header_proof(&self, id: &BlockId<Block>) -> error::Result<(Block::Header, Vec<Vec<u8>>)> {
self.header_proof_with_cht_size(id, cht::SIZE)
}
/// Reads given header and generates CHT-based header proof for CHT of given size.
pub fn header_proof_with_cht_size(&self, id: &BlockId<Block>, cht_size: u64) -> error::Result<(Block::Header, Vec<Vec<u8>>)> {
let proof_error = || error::ErrorKind::Backend(format!("Failed to generate header proof for {:?}", id));
let header = self.header(id)?.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{:?}", id)))?;
let block_num = *header.number();
let cht_num = cht::block_to_cht_number(cht_size, block_num).ok_or_else(proof_error)?;
let cht_start = cht::start_number(cht_size, cht_num);
let headers = (cht_start.as_()..).map(|num| self.block_hash(As::sa(num)).unwrap_or_default());
let proof = cht::build_proof::<Block::Header, Blake2Hasher, _>(cht_size, cht_num, block_num, headers)
.ok_or_else(proof_error)?;
Ok((header, proof))
}
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
/// Get pairs of (block, extrinsic) where key has been changed at given blocks range.
/// Works only for runtimes that are supporting changes tries.
pub fn key_changes(
&self,
first: Block::Hash,
last: Block::Hash,
key: &[u8]
) -> error::Result<Vec<(NumberFor<Block>, u32)>> {
let config = self.changes_trie_config.as_ref();
let storage = self.backend.changes_trie_storage();
let (config, storage) = match (config, storage) {
(Some(config), Some(storage)) => (config, storage),
_ => return Err(error::ErrorKind::ChangesTriesNotSupported.into()),
};
key_changes::<_, Blake2Hasher>(
config,
storage,
self.require_block_number_from_id(&BlockId::Hash(first))?.as_(),
self.require_block_number_from_id(&BlockId::Hash(last))?.as_(),
self.backend.blockchain().info()?.best_number.as_(),
key)
.map_err(|err| error::ErrorKind::ChangesTrieAccessFailed(err).into())
.map(|r| r.into_iter().map(|(b, e)| (As::sa(b), e)).collect())
}
/// Get proof for computation of (block, extrinsic) pairs where key has been changed at given blocks range.
/// `max` is the hash of the last block known to the requester - we can't use changes tries from descendants
/// of this block.
/// Works only for runtimes that are supporting changes tries.
pub fn key_changes_proof(
&self,
first: Block::Hash,
last: Block::Hash,
max: Block::Hash,
key: &[u8]
) -> error::Result<(NumberFor<Block>, Vec<Vec<u8>>)> {
let config = self.changes_trie_config.as_ref();
let storage = self.backend.changes_trie_storage();
let (config, storage) = match (config, storage) {
(Some(config), Some(storage)) => (config, storage),
_ => return Err(error::ErrorKind::ChangesTriesNotSupported.into()),
};
let max_number = ::std::cmp::min(
self.backend.blockchain().info()?.best_number,
self.require_block_number_from_id(&BlockId::Hash(max))?,
);
key_changes_proof::<_, Blake2Hasher>(
config,
storage,
self.require_block_number_from_id(&BlockId::Hash(first))?.as_(),
self.require_block_number_from_id(&BlockId::Hash(last))?.as_(),
max_number.as_(),
key)
.map_err(|err| error::ErrorKind::ChangesTrieAccessFailed(err).into())
.map(|proof| (max_number, proof))
}
/// Create a new block, built on the head of the chain.
pub fn new_block(&self) -> error::Result<block_builder::BlockBuilder<B, E, Block, Blake2Hasher>>
block_builder::BlockBuilder::new(self)
}
/// Create a new block, built on top of `parent`.
pub fn new_block_at(&self, parent: &BlockId<Block>) -> error::Result<block_builder::BlockBuilder<B, E, Block, Blake2Hasher>>
block_builder::BlockBuilder::at_block(parent, &self)
}
/// Set up the native execution environment to call into a native runtime code.
pub fn call_api<A, R>(&self, function: &'static str, args: &A) -> error::Result<R>
where A: Encode, R: Decode
{
self.call_api_at(&BlockId::Number(self.info()?.chain.best_number), function, args)
}
/// Call a runtime function at given block.
pub fn call_api_at<A, R>(&self, at: &BlockId<Block>, function: &'static str, args: &A) -> error::Result<R>
where A: Encode, R: Decode
{
let parent = at;
let header = <<Block as BlockT>::Header as HeaderT>::new(
self.block_number_from_id(&parent)?
.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{:?}", parent)))? + As::sa(1),
Default::default(),
Default::default(),
self.block_hash_from_id(&parent)?
.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{:?}", parent)))?,
Default::default()
);
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
let mut overlay = Default::default();
self.call_at_state(at, "initialise_block", &header, &mut overlay)?;
self.call_at_state(at, function, args, &mut overlay)
}
fn call_at_state<A: Encode, R: Decode>(
&self,
at: &BlockId<Block>,
function: &'static str,
args: &A,
changes: &mut OverlayedChanges
) -> error::Result<R> {
let state = self.state_at(at)?;
let execution_manager = || match self.api_execution_strategy {
ExecutionStrategy::NativeWhenPossible => ExecutionManager::NativeWhenPossible,
ExecutionStrategy::AlwaysWasm => ExecutionManager::AlwaysWasm,
ExecutionStrategy::Both => ExecutionManager::Both(|wasm_result, native_result| {
warn!("Consensus error between wasm and native runtime execution at block {:?}", at);
warn!(" Function {:?}", function);
warn!(" Native result {:?}", native_result);
warn!(" Wasm result {:?}", wasm_result);
wasm_result
}),
};
self.executor.call_at_state(
&state,
changes,
function,
&args.encode(),
execution_manager()
).and_then(|res|
R::decode(&mut &res.0[..])
.ok_or_else(|| Error::from(ErrorKind::CallResultDecode(function)))
)
/// Check a header's justification.
pub fn check_justification(
&self,
header: <Block as BlockT>::Header,
justification: ::bft::UncheckedJustification<Block::Hash>,
) -> error::Result<JustifiedHeader<Block>> {
let parent_hash = header.parent_hash().clone();
let authorities = self.authorities_at(&BlockId::Hash(parent_hash))?;
let just = ::bft::check_justification::<Block>(&authorities[..], parent_hash, justification)
.map_err(|_|
error::ErrorKind::BadJustification(
format!("{}", header.hash())
)
)?;
Ok(JustifiedHeader {
header,
justification: just,
authorities,
})
}
/// Queue a block for import.
pub fn import_block(
&self,
header: JustifiedHeader<Block>,
body: Option<Vec<<Block as BlockT>::Extrinsic>>,
let (header, justification, authorities) = header.into_inner();
let parent_hash = header.parent_hash().clone();
match self.backend.blockchain().status(BlockId::Hash(parent_hash))? {
blockchain::BlockStatus::InChain => {},
blockchain::BlockStatus::Unknown => return Ok(ImportResult::UnknownParent),
}
let _import_lock = self.import_lock.lock();
let height: u64 = header.number().as_();
*self.importing_block.write() = Some(hash);
let result = self.execute_and_import_block(
origin,
hash,
header,
justification,
body,
authorities,
finalized
);
*self.importing_block.write() = None;
telemetry!("block.import";
"height" => height,
"best" => ?hash,
"origin" => ?origin
);
// TODO [ToDr] Optimize and re-use tags from the pool.
fn transaction_tags(&self, at: Block::Hash, body: &Option<Vec<Block::Extrinsic>>) -> error::Result<Vec<TransactionTag>> {
let id = BlockId::Hash(at);
Ok(match body {
None => vec![],
Some(ref extrinsics) => {
let mut tags = vec![];
for tx in extrinsics {
let tx = api::TaggedTransactionQueue::validate_transaction(self, &id, &tx)?;
match tx {
TransactionValidity::Valid(_, _, mut provides, ..) => {
tags.append(&mut provides);
},
// silently ignore invalid extrinsics,
// cause they might just be inherent
_ => {}
}
}
tags
},
})
}
fn execute_and_import_block(
&self,
origin: BlockOrigin,
hash: Block::Hash,
header: Block::Header,
justification: bft::Justification<Block::Hash>,
body: Option<Vec<Block::Extrinsic>>,
authorities: Vec<AuthorityId>,
) -> error::Result<ImportResult> {
let parent_hash = header.parent_hash().clone();
match self.backend.blockchain().status(BlockId::Hash(hash))? {
blockchain::BlockStatus::InChain => return Ok(ImportResult::AlreadyInChain),
blockchain::BlockStatus::Unknown => {},
}
let (last_best, last_best_number) = {
let info = self.backend.blockchain().info()?;
(info.best_hash, info.best_number)
};
// this is a fairly arbitrary choice of where to draw the line on making notifications,
// but the general goal is to only make notifications when we are already fully synced
// and get a new chain head.
let make_notifications = match origin {
BlockOrigin::NetworkBroadcast | BlockOrigin::Own | BlockOrigin::ConsensusBroadcast => true,
BlockOrigin::Genesis | BlockOrigin::NetworkInitialSync | BlockOrigin::File => false,
};
// ensure parent block is finalized to maintain invariant that
// finality is called sequentially.
if finalized {
self.apply_finality(parent_hash, last_best, make_notifications)?;
}
let tags = self.transaction_tags(parent_hash, &body)?;
let mut transaction = self.backend.begin_operation(BlockId::Hash(parent_hash))?;
Svyatoslav Nikolsky
committed
let (storage_update, changes_update, storage_changes) = match transaction.state()? {
Some(transaction_state) => {
let mut overlay = Default::default();
transaction_state,
&mut overlay,
"execute_block",
&<Block as BlockT>::new(header.clone(), body.clone().unwrap_or_default()).encode(),
match (origin, self.block_execution_strategy) {
(BlockOrigin::NetworkInitialSync, _) | (_, ExecutionStrategy::NativeWhenPossible) =>
ExecutionManager::NativeWhenPossible,
(_, ExecutionStrategy::AlwaysWasm) => ExecutionManager::AlwaysWasm,
_ => ExecutionManager::Both(|wasm_result, native_result| {
warn!("Consensus error between wasm and native block execution at block {}", hash);
warn!(" Header {:?}", header);
warn!(" Native result {:?}", native_result);
warn!(" Wasm result {:?}", wasm_result);
telemetry!("block.execute.consensus_failure";
"hash" => ?hash,
"origin" => ?origin,
"header" => ?header
);
wasm_result
}),
},
);
Svyatoslav Nikolsky
committed
let (_, storage_update, changes_update) = r?;
Svyatoslav Nikolsky
committed
(Some(storage_update), Some(changes_update), Some(overlay.into_committed()))
Svyatoslav Nikolsky
committed
None => (None, None, None)
// TODO: non longest-chain rule.
let is_new_best = finalized || header.number() > &last_best_number;
let leaf_state = if finalized {
::backend::NewBlockState::Final
} else if is_new_best {
::backend::NewBlockState::Best
} else {
::backend::NewBlockState::Normal
};
trace!("Imported {}, (#{}), best={}, origin={:?}", hash, header.number(), is_new_best, origin);
let unchecked: bft::UncheckedJustification<_> = justification.uncheck().into();
transaction.set_block_data(
header.clone(),
body,
Some(unchecked.into()),
leaf_state,
)?;
transaction.update_authorities(authorities);
if let Some(storage_update) = storage_update {
transaction.update_storage(storage_update)?;
}
Svyatoslav Nikolsky
committed
if let Some(Some(changes_update)) = changes_update {
transaction.update_changes_trie(changes_update)?;
}
if let Some(storage_changes) = storage_changes {
// TODO [ToDr] How to handle re-orgs? Should we re-emit all storage changes?
self.storage_notifications.lock()
.trigger(&hash, storage_changes);
}
if finalized {
let notification = FinalityNotification::<Block> {
hash,
header: header.clone(),
};
self.finality_notification_sinks.lock()
.retain(|sink| sink.unbounded_send(notification.clone()).is_ok());
}
let notification = BlockImportNotification::<Block> {
hash,
origin,
header,
is_new_best,
self.import_notification_sinks.lock()
.retain(|sink| sink.unbounded_send(notification.clone()).is_ok());
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
/// Finalizes all blocks up to given.
fn apply_finality(&self, block: Block::Hash, best_block: Block::Hash, notify: bool) -> error::Result<()> {
// find tree route from last finalized to given block.
let last_finalized = self.backend.blockchain().last_finalized()?;
if block == last_finalized { return Ok(()) }
let route_from_finalized = ::blockchain::tree_route(
self.backend.blockchain(),
BlockId::Hash(last_finalized),
BlockId::Hash(block),
)?;
if let Some(retracted) = route_from_finalized.retracted().get(0) {
warn!("Safety violation: attempted to revert finalized block {:?} which is not in the \
same chain as last finalized {:?}", retracted, last_finalized);
bail!(error::ErrorKind::NotInFinalizedChain);
}
let route_from_best = ::blockchain::tree_route(
self.backend.blockchain(),
BlockId::Hash(best_block),
BlockId::Hash(block),
)?;
// if the block is not a direct ancestor of the current best chain,
// then some other block is the common ancestor.
if route_from_best.common_block().hash != block {
// TODO: reorganize best block to be the best chain containing
// `block`.
}
for finalize_new in route_from_finalized.enacted() {
self.backend.finalize_block(BlockId::Hash(finalize_new.hash))?;
}
if notify {
// sometimes when syncing, tons of blocks can be finalized at once.
// we'll send notifications spuriously in that case.
const MAX_TO_NOTIFY: usize = 256;
let enacted = route_from_finalized.enacted();
let start = enacted.len() - ::std::cmp::min(enacted.len(), MAX_TO_NOTIFY);
let mut sinks = self.finality_notification_sinks.lock();
for finalized in &enacted[start..] {
let header = self.header(&BlockId::Hash(finalized.hash))?
.expect("header already known to exist in DB because it is indicated in the tree route; qed");
let notification = FinalityNotification {
header,
hash: finalized.hash,
};
sinks.retain(|sink| sink.unbounded_send(notification.clone()).is_ok());
}
}
Ok(())
}
/// Finalize a block. This will implicitly finalize all blocks up to it and
/// fire finality notifications.
///
/// Pass a flag to indicate whether finality notifications should be propagated.
/// This is usually tied to some synchronization state, where we don't send notifications
/// while performing major synchronization work.
pub fn finalize_block(&self, id: BlockId<Block>, notify: bool) -> error::Result<()> {
let last_best = self.backend.blockchain().info()?.best_hash;
let to_finalize_hash = match id {
BlockId::Hash(h) => h,
BlockId::Number(n) => self.backend.blockchain().hash(n)?
.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("No block with number {:?}", n)))?,
};
self.apply_finality(to_finalize_hash, last_best, notify)
}
/// Attempts to revert the chain by `n` blocks. Returns the number of blocks that were
/// successfully reverted.
pub fn revert(&self, n: NumberFor<Block>) -> error::Result<NumberFor<Block>> {
Ok(self.backend.revert(n)?)
}
pub fn info(&self) -> error::Result<ClientInfo<Block>> {
let info = self.backend.blockchain().info().map_err(|e| error::Error::from_blockchain(Box::new(e)))?;
Ok(ClientInfo {
chain: info,
best_queued_hash: None,
best_queued_number: None,
})
}
/// Get block status.
pub fn block_status(&self, id: &BlockId<Block>) -> error::Result<BlockStatus> {
if let BlockId::Hash(ref h) = id {
if self.importing_block.read().as_ref().map_or(false, |importing| h == importing) {
return Ok(BlockStatus::Queued);
}
}
match self.backend.blockchain().header(*id).map_err(|e| error::Error::from_blockchain(Box::new(e)))?.is_some() {
true => Ok(BlockStatus::InChain),
false => Ok(BlockStatus::Unknown),
}
}
/// Get block hash by number.
pub fn block_hash(&self, block_number: <<Block as BlockT>::Header as HeaderT>::Number) -> error::Result<Option<Block::Hash>> {
self.backend.blockchain().hash(block_number)
}
/// Convert an arbitrary block ID into a block hash.
pub fn block_hash_from_id(&self, id: &BlockId<Block>) -> error::Result<Option<Block::Hash>> {
match *id {
BlockId::Hash(h) => Ok(Some(h)),
BlockId::Number(n) => self.block_hash(n),
}
}
/// Convert an arbitrary block ID into a block hash.
pub fn block_number_from_id(&self, id: &BlockId<Block>) -> error::Result<Option<NumberFor<Block>>> {
BlockId::Hash(_) => Ok(self.header(id)?.map(|h| h.number().clone())),
BlockId::Number(n) => Ok(Some(n)),
}
}
/// Convert an arbitrary block ID into a block hash, returning error if the block is unknown.
fn require_block_number_from_id(&self, id: &BlockId<Block>) -> error::Result<NumberFor<Block>> {
self.block_number_from_id(id)
.and_then(|n| n.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{}", id)).into()))
}
pub fn header(&self, id: &BlockId<Block>) -> error::Result<Option<<Block as BlockT>::Header>> {
self.backend.blockchain().header(*id)
}
/// Get block body by id.
pub fn body(&self, id: &BlockId<Block>) -> error::Result<Option<Vec<<Block as BlockT>::Extrinsic>>> {
self.backend.blockchain().body(*id)
}
/// Get block justification set by id.
pub fn justification(&self, id: &BlockId<Block>) -> error::Result<Option<Justification<Block::Hash>>> {
self.backend.blockchain().justification(*id)
}
/// Get full block by id.
pub fn block(&self, id: &BlockId<Block>) -> error::Result<Option<SignedBlock<Block::Header, Block::Extrinsic, Block::Hash>>> {
Ok(match (self.header(id)?, self.body(id)?, self.justification(id)?) {
(Some(header), Some(extrinsics), Some(justification)) =>
Some(SignedBlock { block: RuntimeBlock { header, extrinsics }, justification }),
_ => None,
})
}
/// Get best block header.
pub fn best_block_header(&self) -> error::Result<<Block as BlockT>::Header> {
let info = self.backend.blockchain().info().map_err(|e| error::Error::from_blockchain(Box::new(e)))?;
Ok(self.header(&BlockId::Hash(info.best_hash))?.expect("Best block header must always exist"))
}
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
/// Get the most recent block hash of the best (longest) chains
/// that contain block with the given `target_hash`.
/// If `maybe_max_block_number` is `Some(max_block_number)`
/// the search is limited to block `numbers <= max_block_number`.
/// in other words as if there were no blocks greater `max_block_number`.
/// TODO [snd] possibly implement this on blockchain::Backend and just redirect here
/// Returns `Ok(None)` if `target_hash` is not found in search space.
/// TODO [snd] write down time complexity
pub fn best_containing(&self, target_hash: Block::Hash, maybe_max_number: Option<NumberFor<Block>>) -> error::Result<Option<Block::Hash>> {
let target_header = {
match self.backend.blockchain().header(BlockId::Hash(target_hash))? {
Some(x) => x,
// target not in blockchain
None => { return Ok(None); },
}
};
if let Some(max_number) = maybe_max_number {
// target outside search range
if target_header.number() > &max_number {
return Ok(None);
}
}
let (leaves, best_already_checked) = {
// ensure no blocks are imported during this code block.
// an import could trigger a reorg which could change the canonical chain.
// we depend on the canonical chain staying the same during this code block.
let _import_lock = self.import_lock.lock();
let info = self.backend.blockchain().info()?;
let canon_hash = self.backend.blockchain().hash(*target_header.number())?
.ok_or_else(|| error::Error::from(format!("failed to get hash for block number {}", target_header.number())))?;
if canon_hash == target_hash {
if let Some(max_number) = maybe_max_number {
// something has to guarantee that max_number is in chain
return Ok(Some(self.backend.blockchain().hash(max_number)?.ok_or_else(|| error::Error::from(format!("failed to get hash for block number {}", max_number)))?));
} else {
return Ok(Some(info.best_hash));
}
}
(self.backend.blockchain().leaves()?, info.best_hash)
};
// for each chain. longest chain first. shortest last
for leaf_hash in leaves {
// ignore canonical chain which we already checked above
if leaf_hash == best_already_checked {
continue;
}
// start at the leaf
let mut current_hash = leaf_hash;
// if search is not restricted then the leaf is the best
let mut best_hash = leaf_hash;
// go backwards entering the search space
// waiting until we are <= max_number
if let Some(max_number) = maybe_max_number {
loop {
// TODO [snd] this should be a panic
let current_header = self.backend.blockchain().header(BlockId::Hash(current_hash.clone()))?
.ok_or_else(|| error::Error::from(format!("failed to get header for hash {}", current_hash)))?;
if current_header.number() <= &max_number {
best_hash = current_header.hash();
break;
}
current_hash = *current_header.parent_hash();
}
}
// go backwards through the chain (via parent links)
loop {
// until we find target
if current_hash == target_hash {
return Ok(Some(best_hash));
}
// TODO [snd] this should be a panic
let current_header = self.backend.blockchain().header(BlockId::Hash(current_hash.clone()))?
.ok_or_else(|| error::Error::from(format!("failed to get header for hash {}", current_hash)))?;
// stop search in this chain once we go below the target's block number
if current_header.number() < target_header.number() {
break;
}
current_hash = *current_header.parent_hash();
}
}
unreachable!("this is a bug. `target_hash` is in blockchain but wasn't found following all leaves backwards");
}
impl<B, E, Block> CurrentHeight for Client<B, E, Block> where
B: backend::Backend<Block, Blake2Hasher>,
E: CallExecutor<Block, Blake2Hasher> + Clone,
Block: BlockT,
{
type BlockNumber = <Block::Header as HeaderT>::Number;
fn current_height(&self) -> Self::BlockNumber {
self.backend.blockchain().info().map(|i| i.best_number).unwrap_or_else(|_| Zero::zero())
}
}
impl<B, E, Block> BlockNumberToHash for Client<B, E, Block> where
B: backend::Backend<Block, Blake2Hasher>,
E: CallExecutor<Block, Blake2Hasher> + Clone,
Block: BlockT,
{
type BlockNumber = <Block::Header as HeaderT>::Number;
type Hash = Block::Hash;
fn block_number_to_hash(&self, n: Self::BlockNumber) -> Option<Self::Hash> {
self.block_hash(n).unwrap_or(None)
}
}
impl<B, E, Block> bft::BlockImport<Block> for Client<B, E, Block>
B: backend::Backend<Block, Blake2Hasher>,
E: CallExecutor<Block, Blake2Hasher>,
fn import_block(
&self,
block: Block,
justification: ::bft::Justification<Block::Hash>,
authorities: &[AuthorityId],
asynchronous rob
committed
) -> bool {
let (header, extrinsics) = block.deconstruct();