use std::{collections::HashMap, iter::FromIterator};
use log::{debug, trace};
use sp_runtime::{
traits::{Block as BlockT, Header as _, NumberFor},
Justifications,
};
use crate::{
block_import::{
BlockCheckParams, BlockImport, BlockImportParams, ImportResult, ImportedAux, ImportedState,
JustificationImport, StateAction,
},
metrics::Metrics,
};
pub use basic_queue::BasicQueue;
use sp_consensus::{error::Error as ConsensusError, BlockOrigin, CacheKeyId};
const LOG_TARGET: &str = "sync::import-queue";
pub type DefaultImportQueue<Block, Client> =
BasicQueue<Block, sp_api::TransactionFor<Client, Block>>;
mod basic_queue;
pub mod buffered_link;
pub mod mock;
pub type BoxBlockImport<B, Transaction> =
Box<dyn BlockImport<B, Error = ConsensusError, Transaction = Transaction> + Send + Sync>;
pub type BoxJustificationImport<B> =
Box<dyn JustificationImport<B, Error = ConsensusError> + Send + Sync>;
pub type RuntimeOrigin = libp2p::PeerId;
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct IncomingBlock<B: BlockT> {
pub hash: <B as BlockT>::Hash,
pub header: Option<<B as BlockT>::Header>,
pub body: Option<Vec<<B as BlockT>::Extrinsic>>,
pub indexed_body: Option<Vec<Vec<u8>>>,
pub justifications: Option<Justifications>,
pub origin: Option<RuntimeOrigin>,
pub allow_missing_state: bool,
pub skip_execution: bool,
pub import_existing: bool,
pub state: Option<ImportedState<B>>,
}
#[async_trait::async_trait]
pub trait Verifier<B: BlockT>: Send + Sync {
async fn verify(
&mut self,
block: BlockImportParams<B, ()>,
) -> Result<(BlockImportParams<B, ()>, Option<Vec<(CacheKeyId, Vec<u8>)>>), String>;
}
pub trait ImportQueueService<B: BlockT>: Send {
fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>);
fn import_justifications(
&mut self,
who: RuntimeOrigin,
hash: B::Hash,
number: NumberFor<B>,
justifications: Justifications,
);
}
#[async_trait::async_trait]
pub trait ImportQueue<B: BlockT>: Send {
fn service(&self) -> Box<dyn ImportQueueService<B>>;
fn service_ref(&mut self) -> &mut dyn ImportQueueService<B>;
fn poll_actions(&mut self, cx: &mut futures::task::Context, link: &mut dyn Link<B>);
async fn run(self, link: Box<dyn Link<B>>);
}
pub trait Link<B: BlockT>: Send {
fn blocks_processed(
&mut self,
_imported: usize,
_count: usize,
_results: Vec<(BlockImportResult<B>, B::Hash)>,
) {
}
fn justification_imported(
&mut self,
_who: RuntimeOrigin,
_hash: &B::Hash,
_number: NumberFor<B>,
_success: bool,
) {
}
fn request_justification(&mut self, _hash: &B::Hash, _number: NumberFor<B>) {}
}
#[derive(Debug, PartialEq)]
pub enum BlockImportStatus<N: std::fmt::Debug + PartialEq> {
ImportedKnown(N, Option<RuntimeOrigin>),
ImportedUnknown(N, ImportedAux, Option<RuntimeOrigin>),
}
impl<N: std::fmt::Debug + PartialEq> BlockImportStatus<N> {
pub fn number(&self) -> &N {
match self {
BlockImportStatus::ImportedKnown(n, _) |
BlockImportStatus::ImportedUnknown(n, _, _) => n,
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum BlockImportError {
#[error("block is missing a header (origin = {0:?})")]
IncompleteHeader(Option<RuntimeOrigin>),
#[error("block verification failed (origin = {0:?}): {1}")]
VerificationFailed(Option<RuntimeOrigin>, String),
#[error("bad block (origin = {0:?})")]
BadBlock(Option<RuntimeOrigin>),
#[error("block is missing parent state")]
MissingState,
#[error("block has an unknown parent")]
UnknownParent,
#[error("import has been cancelled")]
Cancelled,
#[error("consensus error: {0}")]
Other(ConsensusError),
}
type BlockImportResult<B> = Result<BlockImportStatus<NumberFor<B>>, BlockImportError>;
pub async fn import_single_block<B: BlockT, V: Verifier<B>, Transaction: Send + 'static>(
import_handle: &mut impl BlockImport<B, Transaction = Transaction, Error = ConsensusError>,
block_origin: BlockOrigin,
block: IncomingBlock<B>,
verifier: &mut V,
) -> BlockImportResult<B> {
import_single_block_metered(import_handle, block_origin, block, verifier, None).await
}
pub(crate) async fn import_single_block_metered<
B: BlockT,
V: Verifier<B>,
Transaction: Send + 'static,
>(
import_handle: &mut impl BlockImport<B, Transaction = Transaction, Error = ConsensusError>,
block_origin: BlockOrigin,
block: IncomingBlock<B>,
verifier: &mut V,
metrics: Option<Metrics>,
) -> BlockImportResult<B> {
let peer = block.origin;
let (header, justifications) = match (block.header, block.justifications) {
(Some(header), justifications) => (header, justifications),
(None, _) => {
if let Some(ref peer) = peer {
debug!(target: LOG_TARGET, "Header {} was not provided by {} ", block.hash, peer);
} else {
debug!(target: LOG_TARGET, "Header {} was not provided ", block.hash);
}
return Err(BlockImportError::IncompleteHeader(peer))
},
};
trace!(target: LOG_TARGET, "Header {} has {:?} logs", block.hash, header.digest().logs().len());
let number = *header.number();
let hash = block.hash;
let parent_hash = *header.parent_hash();
let import_handler = |import| match import {
Ok(ImportResult::AlreadyInChain) => {
trace!(target: LOG_TARGET, "Block already in chain {}: {:?}", number, hash);
Ok(BlockImportStatus::ImportedKnown(number, peer))
},
Ok(ImportResult::Imported(aux)) =>
Ok(BlockImportStatus::ImportedUnknown(number, aux, peer)),
Ok(ImportResult::MissingState) => {
debug!(
target: LOG_TARGET,
"Parent state is missing for {}: {:?}, parent: {:?}", number, hash, parent_hash
);
Err(BlockImportError::MissingState)
},
Ok(ImportResult::UnknownParent) => {
debug!(
target: LOG_TARGET,
"Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent_hash
);
Err(BlockImportError::UnknownParent)
},
Ok(ImportResult::KnownBad) => {
debug!(target: LOG_TARGET, "Peer gave us a bad block {}: {:?}", number, hash);
Err(BlockImportError::BadBlock(peer))
},
Err(e) => {
debug!(target: LOG_TARGET, "Error importing block {}: {:?}: {}", number, hash, e);
Err(BlockImportError::Other(e))
},
};
match import_handler(
import_handle
.check_block(BlockCheckParams {
hash,
number,
parent_hash,
allow_missing_state: block.allow_missing_state,
import_existing: block.import_existing,
allow_missing_parent: block.state.is_some(),
})
.await,
)? {
BlockImportStatus::ImportedUnknown { .. } => (),
r => return Ok(r), }
let started = std::time::Instant::now();
let mut import_block = BlockImportParams::new(block_origin, header);
import_block.body = block.body;
import_block.justifications = justifications;
import_block.post_hash = Some(hash);
import_block.import_existing = block.import_existing;
import_block.indexed_body = block.indexed_body;
if let Some(state) = block.state {
let changes = crate::block_import::StorageChanges::Import(state);
import_block.state_action = StateAction::ApplyChanges(changes);
} else if block.skip_execution {
import_block.state_action = StateAction::Skip;
} else if block.allow_missing_state {
import_block.state_action = StateAction::ExecuteIfPossible;
}
let (import_block, maybe_keys) = verifier.verify(import_block).await.map_err(|msg| {
if let Some(ref peer) = peer {
trace!(
target: LOG_TARGET,
"Verifying {}({}) from {} failed: {}",
number,
hash,
peer,
msg
);
} else {
trace!(target: LOG_TARGET, "Verifying {}({}) failed: {}", number, hash, msg);
}
if let Some(metrics) = metrics.as_ref() {
metrics.report_verification(false, started.elapsed());
}
BlockImportError::VerificationFailed(peer, msg)
})?;
if let Some(metrics) = metrics.as_ref() {
metrics.report_verification(true, started.elapsed());
}
let cache = HashMap::from_iter(maybe_keys.unwrap_or_default());
let import_block = import_block.clear_storage_changes_and_mutate();
let imported = import_handle.import_block(import_block, cache).await;
if let Some(metrics) = metrics.as_ref() {
metrics.report_verification_and_import(started.elapsed());
}
import_handler(imported)
}