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
144#[derive(Debug, PartialEq)]
146pub enum JustificationImportResult {
147 Success,
149
150 Failure,
152
153 OutdatedJustification,
155}
156
157pub trait Link<B: BlockT>: Send + Sync {
160 fn blocks_processed(
162 &self,
163 _imported: usize,
164 _count: usize,
165 _results: Vec<(BlockImportResult<B>, B::Hash)>,
166 ) {
167 }
168
169 fn justification_imported(
171 &self,
172 _who: RuntimeOrigin,
173 _hash: &B::Hash,
174 _number: NumberFor<B>,
175 _import_result: JustificationImportResult,
176 ) {
177 }
178
179 fn request_justification(&self, _hash: &B::Hash, _number: NumberFor<B>) {}
181}
182
183#[derive(Debug, PartialEq)]
185pub enum BlockImportStatus<BlockNumber: fmt::Debug + PartialEq> {
186 ImportedKnown(BlockNumber, Option<RuntimeOrigin>),
188 ImportedUnknown(BlockNumber, ImportedAux, Option<RuntimeOrigin>),
190}
191
192impl<BlockNumber: fmt::Debug + PartialEq> BlockImportStatus<BlockNumber> {
193 pub fn number(&self) -> &BlockNumber {
195 match self {
196 BlockImportStatus::ImportedKnown(n, _) |
197 BlockImportStatus::ImportedUnknown(n, _, _) => n,
198 }
199 }
200}
201
202#[derive(Debug, thiserror::Error)]
204pub enum BlockImportError {
205 #[error("block is missing a header (origin = {0:?})")]
207 IncompleteHeader(Option<RuntimeOrigin>),
208
209 #[error("block verification failed (origin = {0:?}): {1}")]
211 VerificationFailed(Option<RuntimeOrigin>, String),
212
213 #[error("bad block (origin = {0:?})")]
215 BadBlock(Option<RuntimeOrigin>),
216
217 #[error("block is missing parent state")]
219 MissingState,
220
221 #[error("block has an unknown parent")]
223 UnknownParent,
224
225 #[error("import has been cancelled")]
227 Cancelled,
228
229 #[error("consensus error: {0}")]
231 Other(ConsensusError),
232}
233
234type BlockImportResult<B> = Result<BlockImportStatus<NumberFor<B>>, BlockImportError>;
235
236pub async fn import_single_block<B: BlockT, V: Verifier<B>>(
238 import_handle: &mut impl BlockImport<B, Error = ConsensusError>,
239 block_origin: BlockOrigin,
240 block: IncomingBlock<B>,
241 verifier: &V,
242) -> BlockImportResult<B> {
243 match verify_single_block_metered(import_handle, block_origin, block, verifier, None).await? {
244 SingleBlockVerificationOutcome::Imported(import_status) => Ok(import_status),
245 SingleBlockVerificationOutcome::Verified(import_parameters) =>
246 import_single_block_metered(import_handle, import_parameters, None).await,
247 }
248}
249
250fn import_handler<Block>(
251 number: NumberFor<Block>,
252 hash: Block::Hash,
253 parent_hash: Block::Hash,
254 block_origin: Option<RuntimeOrigin>,
255 import: Result<ImportResult, ConsensusError>,
256) -> Result<BlockImportStatus<NumberFor<Block>>, BlockImportError>
257where
258 Block: BlockT,
259{
260 match import {
261 Ok(ImportResult::AlreadyInChain) => {
262 trace!(target: LOG_TARGET, "Block already in chain {}: {:?}", number, hash);
263 Ok(BlockImportStatus::ImportedKnown(number, block_origin))
264 },
265 Ok(ImportResult::Imported(aux)) =>
266 Ok(BlockImportStatus::ImportedUnknown(number, aux, block_origin)),
267 Ok(ImportResult::MissingState) => {
268 debug!(
269 target: LOG_TARGET,
270 "Parent state is missing for {}: {:?}, parent: {:?}", number, hash, parent_hash
271 );
272 Err(BlockImportError::MissingState)
273 },
274 Ok(ImportResult::UnknownParent) => {
275 debug!(
276 target: LOG_TARGET,
277 "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent_hash
278 );
279 Err(BlockImportError::UnknownParent)
280 },
281 Ok(ImportResult::KnownBad) => {
282 debug!(target: LOG_TARGET, "Peer gave us a bad block {}: {:?}", number, hash);
283 Err(BlockImportError::BadBlock(block_origin))
284 },
285 Err(e) => {
286 debug!(target: LOG_TARGET, "Error importing block {}: {:?}: {}", number, hash, e);
287 Err(BlockImportError::Other(e))
288 },
289 }
290}
291
292pub(crate) enum SingleBlockVerificationOutcome<Block: BlockT> {
293 Imported(BlockImportStatus<NumberFor<Block>>),
295 Verified(SingleBlockImportParameters<Block>),
297}
298
299pub(crate) struct SingleBlockImportParameters<Block: BlockT> {
300 import_block: BlockImportParams<Block>,
301 hash: Block::Hash,
302 block_origin: Option<RuntimeOrigin>,
303 verification_time: Duration,
304}
305
306pub(crate) async fn verify_single_block_metered<B: BlockT, V: Verifier<B>>(
308 import_handle: &impl BlockImport<B, Error = ConsensusError>,
309 block_origin: BlockOrigin,
310 block: IncomingBlock<B>,
311 verifier: &V,
312 metrics: Option<&Metrics>,
313) -> Result<SingleBlockVerificationOutcome<B>, BlockImportError> {
314 let peer = block.origin;
315 let justifications = block.justifications;
316
317 let Some(header) = block.header else {
318 if let Some(ref peer) = peer {
319 debug!(target: LOG_TARGET, "Header {} was not provided by {peer} ", block.hash);
320 } else {
321 debug!(target: LOG_TARGET, "Header {} was not provided ", block.hash);
322 }
323 return Err(BlockImportError::IncompleteHeader(peer))
324 };
325
326 trace!(target: LOG_TARGET, "Header {} has {:?} logs", block.hash, header.digest().logs().len());
327
328 let number = *header.number();
329 let hash = block.hash;
330 let parent_hash = *header.parent_hash();
331
332 match import_handler::<B>(
333 number,
334 hash,
335 parent_hash,
336 peer,
337 import_handle
338 .check_block(BlockCheckParams {
339 hash,
340 number,
341 parent_hash,
342 allow_missing_state: block.allow_missing_state,
343 import_existing: block.import_existing,
344 allow_missing_parent: block.state.is_some(),
345 })
346 .await,
347 )? {
348 BlockImportStatus::ImportedUnknown { .. } => (),
349 r => {
350 return Ok(SingleBlockVerificationOutcome::Imported(r))
352 },
353 }
354
355 let started = Instant::now();
356
357 let mut import_block = BlockImportParams::new(block_origin, header);
358 import_block.body = block.body;
359 import_block.justifications = justifications;
360 import_block.post_hash = Some(hash);
361 import_block.import_existing = block.import_existing;
362 import_block.indexed_body = block.indexed_body;
363
364 if let Some(state) = block.state {
365 let changes = crate::block_import::StorageChanges::Import(state);
366 import_block.state_action = StateAction::ApplyChanges(changes);
367 } else if block.skip_execution {
368 import_block.state_action = StateAction::Skip;
369 } else if block.allow_missing_state {
370 import_block.state_action = StateAction::ExecuteIfPossible;
371 }
372
373 let import_block = verifier.verify(import_block).await.map_err(|msg| {
374 if let Some(ref peer) = peer {
375 trace!(
376 target: LOG_TARGET,
377 "Verifying {}({}) from {} failed: {}",
378 number,
379 hash,
380 peer,
381 msg
382 );
383 } else {
384 trace!(target: LOG_TARGET, "Verifying {}({}) failed: {}", number, hash, msg);
385 }
386 if let Some(metrics) = metrics {
387 metrics.report_verification(false, started.elapsed());
388 }
389 BlockImportError::VerificationFailed(peer, msg)
390 })?;
391
392 let verification_time = started.elapsed();
393 if let Some(metrics) = metrics {
394 metrics.report_verification(true, verification_time);
395 }
396
397 Ok(SingleBlockVerificationOutcome::Verified(SingleBlockImportParameters {
398 import_block,
399 hash,
400 block_origin: peer,
401 verification_time,
402 }))
403}
404
405pub(crate) async fn import_single_block_metered<Block: BlockT>(
406 import_handle: &mut impl BlockImport<Block, Error = ConsensusError>,
407 import_parameters: SingleBlockImportParameters<Block>,
408 metrics: Option<&Metrics>,
409) -> BlockImportResult<Block> {
410 let started = Instant::now();
411
412 let SingleBlockImportParameters { import_block, hash, block_origin, verification_time } =
413 import_parameters;
414
415 let number = *import_block.header.number();
416 let parent_hash = *import_block.header.parent_hash();
417
418 let imported = import_handle.import_block(import_block).await;
419 if let Some(metrics) = metrics {
420 metrics.report_verification_and_import(started.elapsed() + verification_time);
421 }
422
423 import_handler::<Block>(number, hash, parent_hash, block_origin, imported)
424}