1use log::{debug, trace};
31use std::{
32 fmt,
33 time::{Duration, Instant},
34};
35
36use sp_consensus::{error::Error as ConsensusError, BlockOrigin};
37use sp_runtime::{
38 traits::{Block as BlockT, Header as _, NumberFor},
39 Justifications,
40};
41
42use crate::{
43 block_import::{
44 BlockCheckParams, BlockImport, BlockImportParams, ImportResult, ImportedAux, ImportedState,
45 JustificationImport, StateAction,
46 },
47 metrics::Metrics,
48};
49
50pub use basic_queue::BasicQueue;
51
52const LOG_TARGET: &str = "sync::import-queue";
53
54pub type DefaultImportQueue<Block> = BasicQueue<Block>;
58
59mod basic_queue;
60pub mod buffered_link;
61pub mod mock;
62
63pub type BoxBlockImport<B> = Box<dyn BlockImport<B, Error = ConsensusError> + Send + Sync>;
65
66pub type BoxJustificationImport<B> =
68 Box<dyn JustificationImport<B, Error = ConsensusError> + Send + Sync>;
69
70pub type RuntimeOrigin = sc_network_types::PeerId;
72
73#[derive(Debug, PartialEq, Eq, Clone)]
75pub struct IncomingBlock<B: BlockT> {
76 pub hash: <B as BlockT>::Hash,
78 pub header: Option<<B as BlockT>::Header>,
80 pub body: Option<Vec<<B as BlockT>::Extrinsic>>,
82 pub indexed_body: Option<Vec<Vec<u8>>>,
84 pub justifications: Option<Justifications>,
86 pub origin: Option<RuntimeOrigin>,
88 pub allow_missing_state: bool,
90 pub skip_execution: bool,
92 pub import_existing: bool,
94 pub state: Option<ImportedState<B>>,
96}
97
98#[async_trait::async_trait]
100pub trait Verifier<B: BlockT>: Send + Sync {
101 async fn verify(&self, block: BlockImportParams<B>) -> Result<BlockImportParams<B>, String>;
104}
105
106pub trait ImportQueueService<B: BlockT>: Send {
110 fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>);
113
114 fn import_justifications(
116 &mut self,
117 who: RuntimeOrigin,
118 hash: B::Hash,
119 number: NumberFor<B>,
120 justifications: Justifications,
121 );
122}
123
124#[async_trait::async_trait]
125pub trait ImportQueue<B: BlockT>: Send {
126 fn service(&self) -> Box<dyn ImportQueueService<B>>;
128
129 fn service_ref(&mut self) -> &mut dyn ImportQueueService<B>;
131
132 fn poll_actions(&mut self, cx: &mut futures::task::Context, link: &dyn Link<B>);
136
137 async fn run(self, link: &dyn Link<B>);
142}
143
144pub trait Link<B: BlockT>: Send + Sync {
147 fn blocks_processed(
149 &self,
150 _imported: usize,
151 _count: usize,
152 _results: Vec<(BlockImportResult<B>, B::Hash)>,
153 ) {
154 }
155
156 fn justification_imported(
158 &self,
159 _who: RuntimeOrigin,
160 _hash: &B::Hash,
161 _number: NumberFor<B>,
162 _success: bool,
163 ) {
164 }
165
166 fn request_justification(&self, _hash: &B::Hash, _number: NumberFor<B>) {}
168}
169
170#[derive(Debug, PartialEq)]
172pub enum BlockImportStatus<BlockNumber: fmt::Debug + PartialEq> {
173 ImportedKnown(BlockNumber, Option<RuntimeOrigin>),
175 ImportedUnknown(BlockNumber, ImportedAux, Option<RuntimeOrigin>),
177}
178
179impl<BlockNumber: fmt::Debug + PartialEq> BlockImportStatus<BlockNumber> {
180 pub fn number(&self) -> &BlockNumber {
182 match self {
183 BlockImportStatus::ImportedKnown(n, _) |
184 BlockImportStatus::ImportedUnknown(n, _, _) => n,
185 }
186 }
187}
188
189#[derive(Debug, thiserror::Error)]
191pub enum BlockImportError {
192 #[error("block is missing a header (origin = {0:?})")]
194 IncompleteHeader(Option<RuntimeOrigin>),
195
196 #[error("block verification failed (origin = {0:?}): {1}")]
198 VerificationFailed(Option<RuntimeOrigin>, String),
199
200 #[error("bad block (origin = {0:?})")]
202 BadBlock(Option<RuntimeOrigin>),
203
204 #[error("block is missing parent state")]
206 MissingState,
207
208 #[error("block has an unknown parent")]
210 UnknownParent,
211
212 #[error("import has been cancelled")]
214 Cancelled,
215
216 #[error("consensus error: {0}")]
218 Other(ConsensusError),
219}
220
221type BlockImportResult<B> = Result<BlockImportStatus<NumberFor<B>>, BlockImportError>;
222
223pub async fn import_single_block<B: BlockT, V: Verifier<B>>(
225 import_handle: &mut impl BlockImport<B, Error = ConsensusError>,
226 block_origin: BlockOrigin,
227 block: IncomingBlock<B>,
228 verifier: &V,
229) -> BlockImportResult<B> {
230 match verify_single_block_metered(import_handle, block_origin, block, verifier, None).await? {
231 SingleBlockVerificationOutcome::Imported(import_status) => Ok(import_status),
232 SingleBlockVerificationOutcome::Verified(import_parameters) =>
233 import_single_block_metered(import_handle, import_parameters, None).await,
234 }
235}
236
237fn import_handler<Block>(
238 number: NumberFor<Block>,
239 hash: Block::Hash,
240 parent_hash: Block::Hash,
241 block_origin: Option<RuntimeOrigin>,
242 import: Result<ImportResult, ConsensusError>,
243) -> Result<BlockImportStatus<NumberFor<Block>>, BlockImportError>
244where
245 Block: BlockT,
246{
247 match import {
248 Ok(ImportResult::AlreadyInChain) => {
249 trace!(target: LOG_TARGET, "Block already in chain {}: {:?}", number, hash);
250 Ok(BlockImportStatus::ImportedKnown(number, block_origin))
251 },
252 Ok(ImportResult::Imported(aux)) =>
253 Ok(BlockImportStatus::ImportedUnknown(number, aux, block_origin)),
254 Ok(ImportResult::MissingState) => {
255 debug!(
256 target: LOG_TARGET,
257 "Parent state is missing for {}: {:?}, parent: {:?}", number, hash, parent_hash
258 );
259 Err(BlockImportError::MissingState)
260 },
261 Ok(ImportResult::UnknownParent) => {
262 debug!(
263 target: LOG_TARGET,
264 "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent_hash
265 );
266 Err(BlockImportError::UnknownParent)
267 },
268 Ok(ImportResult::KnownBad) => {
269 debug!(target: LOG_TARGET, "Peer gave us a bad block {}: {:?}", number, hash);
270 Err(BlockImportError::BadBlock(block_origin))
271 },
272 Err(e) => {
273 debug!(target: LOG_TARGET, "Error importing block {}: {:?}: {}", number, hash, e);
274 Err(BlockImportError::Other(e))
275 },
276 }
277}
278
279pub(crate) enum SingleBlockVerificationOutcome<Block: BlockT> {
280 Imported(BlockImportStatus<NumberFor<Block>>),
282 Verified(SingleBlockImportParameters<Block>),
284}
285
286pub(crate) struct SingleBlockImportParameters<Block: BlockT> {
287 import_block: BlockImportParams<Block>,
288 hash: Block::Hash,
289 block_origin: Option<RuntimeOrigin>,
290 verification_time: Duration,
291}
292
293pub(crate) async fn verify_single_block_metered<B: BlockT, V: Verifier<B>>(
295 import_handle: &impl BlockImport<B, Error = ConsensusError>,
296 block_origin: BlockOrigin,
297 block: IncomingBlock<B>,
298 verifier: &V,
299 metrics: Option<&Metrics>,
300) -> Result<SingleBlockVerificationOutcome<B>, BlockImportError> {
301 let peer = block.origin;
302 let justifications = block.justifications;
303
304 let Some(header) = block.header else {
305 if let Some(ref peer) = peer {
306 debug!(target: LOG_TARGET, "Header {} was not provided by {peer} ", block.hash);
307 } else {
308 debug!(target: LOG_TARGET, "Header {} was not provided ", block.hash);
309 }
310 return Err(BlockImportError::IncompleteHeader(peer))
311 };
312
313 trace!(target: LOG_TARGET, "Header {} has {:?} logs", block.hash, header.digest().logs().len());
314
315 let number = *header.number();
316 let hash = block.hash;
317 let parent_hash = *header.parent_hash();
318
319 match import_handler::<B>(
320 number,
321 hash,
322 parent_hash,
323 peer,
324 import_handle
325 .check_block(BlockCheckParams {
326 hash,
327 number,
328 parent_hash,
329 allow_missing_state: block.allow_missing_state,
330 import_existing: block.import_existing,
331 allow_missing_parent: block.state.is_some(),
332 })
333 .await,
334 )? {
335 BlockImportStatus::ImportedUnknown { .. } => (),
336 r => {
337 return Ok(SingleBlockVerificationOutcome::Imported(r))
339 },
340 }
341
342 let started = Instant::now();
343
344 let mut import_block = BlockImportParams::new(block_origin, header);
345 import_block.body = block.body;
346 import_block.justifications = justifications;
347 import_block.post_hash = Some(hash);
348 import_block.import_existing = block.import_existing;
349 import_block.indexed_body = block.indexed_body;
350
351 if let Some(state) = block.state {
352 let changes = crate::block_import::StorageChanges::Import(state);
353 import_block.state_action = StateAction::ApplyChanges(changes);
354 } else if block.skip_execution {
355 import_block.state_action = StateAction::Skip;
356 } else if block.allow_missing_state {
357 import_block.state_action = StateAction::ExecuteIfPossible;
358 }
359
360 let import_block = verifier.verify(import_block).await.map_err(|msg| {
361 if let Some(ref peer) = peer {
362 trace!(
363 target: LOG_TARGET,
364 "Verifying {}({}) from {} failed: {}",
365 number,
366 hash,
367 peer,
368 msg
369 );
370 } else {
371 trace!(target: LOG_TARGET, "Verifying {}({}) failed: {}", number, hash, msg);
372 }
373 if let Some(metrics) = metrics {
374 metrics.report_verification(false, started.elapsed());
375 }
376 BlockImportError::VerificationFailed(peer, msg)
377 })?;
378
379 let verification_time = started.elapsed();
380 if let Some(metrics) = metrics {
381 metrics.report_verification(true, verification_time);
382 }
383
384 Ok(SingleBlockVerificationOutcome::Verified(SingleBlockImportParameters {
385 import_block,
386 hash,
387 block_origin: peer,
388 verification_time,
389 }))
390}
391
392pub(crate) async fn import_single_block_metered<Block: BlockT>(
393 import_handle: &mut impl BlockImport<Block, Error = ConsensusError>,
394 import_parameters: SingleBlockImportParameters<Block>,
395 metrics: Option<&Metrics>,
396) -> BlockImportResult<Block> {
397 let started = Instant::now();
398
399 let SingleBlockImportParameters { import_block, hash, block_origin, verification_time } =
400 import_parameters;
401
402 let number = *import_block.header.number();
403 let parent_hash = *import_block.header.parent_hash();
404
405 let imported = import_handle.import_block(import_block).await;
406 if let Some(metrics) = metrics {
407 metrics.report_verification_and_import(started.elapsed() + verification_time);
408 }
409
410 import_handler::<Block>(number, hash, parent_hash, block_origin, imported)
411}