referrerpolicy=no-referrer-when-downgrade

sc_consensus/
import_queue.rs

1// This file is part of Substrate.
2
3// Copyright (C) Parity Technologies (UK) Ltd.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! Import Queue primitive: something which can verify and import blocks.
20//!
21//! This serves as an intermediate and abstracted step between synchronization
22//! and import. Each mode of consensus will have its own requirements for block
23//! verification. Some algorithms can verify in parallel, while others only
24//! sequentially.
25//!
26//! The `ImportQueue` trait allows such verification strategies to be
27//! instantiated. The `BasicQueue` and `BasicVerifier` traits allow serial
28//! queues to be instantiated simply.
29
30use 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
54/// A commonly-used Import Queue type.
55///
56/// This defines the transaction type of the `BasicQueue` to be the transaction type for a client.
57pub type DefaultImportQueue<Block> = BasicQueue<Block>;
58
59mod basic_queue;
60pub mod buffered_link;
61pub mod mock;
62
63/// Shared block import struct used by the queue.
64pub type BoxBlockImport<B> = Box<dyn BlockImport<B, Error = ConsensusError> + Send + Sync>;
65
66/// Shared justification import struct used by the queue.
67pub type BoxJustificationImport<B> =
68	Box<dyn JustificationImport<B, Error = ConsensusError> + Send + Sync>;
69
70/// Maps to the RuntimeOrigin used by the network.
71pub type RuntimeOrigin = sc_network_types::PeerId;
72
73/// Block data used by the queue.
74#[derive(Debug, PartialEq, Eq, Clone)]
75pub struct IncomingBlock<B: BlockT> {
76	/// Block header hash.
77	pub hash: <B as BlockT>::Hash,
78	/// Block header if requested.
79	pub header: Option<<B as BlockT>::Header>,
80	/// Block body if requested.
81	pub body: Option<Vec<<B as BlockT>::Extrinsic>>,
82	/// Indexed block body if requested.
83	pub indexed_body: Option<Vec<Vec<u8>>>,
84	/// Justification(s) if requested.
85	pub justifications: Option<Justifications>,
86	/// The peer, we received this from
87	pub origin: Option<RuntimeOrigin>,
88	/// Allow importing the block skipping state verification if parent state is missing.
89	pub allow_missing_state: bool,
90	/// Skip block execution and state verification.
91	pub skip_execution: bool,
92	/// Re-validate existing block.
93	pub import_existing: bool,
94	/// Do not compute new state, but rather set it to the given set.
95	pub state: Option<ImportedState<B>>,
96}
97
98/// Verify a justification of a block
99#[async_trait::async_trait]
100pub trait Verifier<B: BlockT>: Send + Sync {
101	/// Verify the given block data and return the `BlockImportParams` to
102	/// continue the block import process.
103	async fn verify(&self, block: BlockImportParams<B>) -> Result<BlockImportParams<B>, String>;
104}
105
106/// Blocks import queue API.
107///
108/// The `import_*` methods can be called in order to send elements for the import queue to verify.
109pub trait ImportQueueService<B: BlockT>: Send {
110	/// Import a bunch of blocks, every next block must be an ancestor of the previous block in the
111	/// list.
112	fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>);
113
114	/// Import block justifications.
115	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	/// Get a copy of the handle to [`ImportQueueService`].
127	fn service(&self) -> Box<dyn ImportQueueService<B>>;
128
129	/// Get a reference to the handle to [`ImportQueueService`].
130	fn service_ref(&mut self) -> &mut dyn ImportQueueService<B>;
131
132	/// This method should behave in a way similar to `Future::poll`. It can register the current
133	/// task and notify later when more actions are ready to be polled. To continue the comparison,
134	/// it is as if this method always returned `Poll::Pending`.
135	fn poll_actions(&mut self, cx: &mut futures::task::Context, link: &dyn Link<B>);
136
137	/// Start asynchronous runner for import queue.
138	///
139	/// Takes an object implementing [`Link`] which allows the import queue to
140	/// influence the synchronization process.
141	async fn run(self, link: &dyn Link<B>);
142}
143
144/// Hooks that the verification queue can use to influence the synchronization
145/// algorithm.
146pub trait Link<B: BlockT>: Send + Sync {
147	/// Batch of blocks imported, with or without error.
148	fn blocks_processed(
149		&self,
150		_imported: usize,
151		_count: usize,
152		_results: Vec<(BlockImportResult<B>, B::Hash)>,
153	) {
154	}
155
156	/// Justification import result.
157	fn justification_imported(
158		&self,
159		_who: RuntimeOrigin,
160		_hash: &B::Hash,
161		_number: NumberFor<B>,
162		_success: bool,
163	) {
164	}
165
166	/// Request a justification for the given block.
167	fn request_justification(&self, _hash: &B::Hash, _number: NumberFor<B>) {}
168}
169
170/// Block import successful result.
171#[derive(Debug, PartialEq)]
172pub enum BlockImportStatus<BlockNumber: fmt::Debug + PartialEq> {
173	/// Imported known block.
174	ImportedKnown(BlockNumber, Option<RuntimeOrigin>),
175	/// Imported unknown block.
176	ImportedUnknown(BlockNumber, ImportedAux, Option<RuntimeOrigin>),
177}
178
179impl<BlockNumber: fmt::Debug + PartialEq> BlockImportStatus<BlockNumber> {
180	/// Returns the imported block number.
181	pub fn number(&self) -> &BlockNumber {
182		match self {
183			BlockImportStatus::ImportedKnown(n, _) |
184			BlockImportStatus::ImportedUnknown(n, _, _) => n,
185		}
186	}
187}
188
189/// Block import error.
190#[derive(Debug, thiserror::Error)]
191pub enum BlockImportError {
192	/// Block missed header, can't be imported
193	#[error("block is missing a header (origin = {0:?})")]
194	IncompleteHeader(Option<RuntimeOrigin>),
195
196	/// Block verification failed, can't be imported
197	#[error("block verification failed (origin = {0:?}): {1}")]
198	VerificationFailed(Option<RuntimeOrigin>, String),
199
200	/// Block is known to be Bad
201	#[error("bad block (origin = {0:?})")]
202	BadBlock(Option<RuntimeOrigin>),
203
204	/// Parent state is missing.
205	#[error("block is missing parent state")]
206	MissingState,
207
208	/// Block has an unknown parent
209	#[error("block has an unknown parent")]
210	UnknownParent,
211
212	/// Block import has been cancelled. This can happen if the parent block fails to be imported.
213	#[error("import has been cancelled")]
214	Cancelled,
215
216	/// Other error.
217	#[error("consensus error: {0}")]
218	Other(ConsensusError),
219}
220
221type BlockImportResult<B> = Result<BlockImportStatus<NumberFor<B>>, BlockImportError>;
222
223/// Single block import function.
224pub 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	/// Block is already imported.
281	Imported(BlockImportStatus<NumberFor<Block>>),
282	/// Block is verified, but needs to be imported.
283	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
293/// Single block import function with metering.
294pub(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			// Any other successful result means that the block is already imported.
338			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}