referrerpolicy=no-referrer-when-downgrade

sc_network_sync/
block_request_handler.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Substrate.
3// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
4
5// Substrate is free software: you can redistribute it and/or modify
6// it under the terms of the GNU General Public License as published by
7// the Free Software Foundation, either version 3 of the License, or
8// (at your option) any later version.
9
10// Substrate is distributed in the hope that it will be useful,
11// but WITHOUT ANY WARRANTY; without even the implied warranty of
12// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13// GNU General Public License for more details.
14
15// You should have received a copy of the GNU General Public License
16// along with Substrate. If not, see <https://www.gnu.org/licenses/>.
17
18//! Helper for handling (i.e. answering) block requests from a remote peer via the
19//! `crate::request_responses::RequestResponsesBehaviour`.
20
21use crate::{
22	block_relay_protocol::{BlockDownloader, BlockRelayParams, BlockResponseError, BlockServer},
23	schema::v1::{
24		block_request::FromBlock as FromBlockSchema, BlockRequest as BlockRequestSchema,
25		BlockResponse as BlockResponseSchema, BlockResponse, Direction,
26	},
27	service::network::NetworkServiceHandle,
28	LOG_TARGET,
29};
30
31use codec::{Decode, DecodeAll, Encode};
32use futures::{channel::oneshot, stream::StreamExt};
33use log::debug;
34use prost::Message;
35use schnellru::{ByLength, LruMap};
36
37use sc_client_api::BlockBackend;
38use sc_network::{
39	config::ProtocolId,
40	request_responses::{IfDisconnected, IncomingRequest, OutgoingResponse, RequestFailure},
41	service::traits::RequestResponseConfig,
42	types::ProtocolName,
43	NetworkBackend, MAX_RESPONSE_SIZE,
44};
45use sc_network_common::sync::message::{BlockAttributes, BlockData, BlockRequest, FromBlock};
46use sc_network_types::PeerId;
47use sp_blockchain::HeaderBackend;
48use sp_runtime::{
49	generic::BlockId,
50	traits::{Block as BlockT, Header, One, Zero},
51};
52
53use std::{
54	cmp::min,
55	hash::{Hash, Hasher},
56	sync::Arc,
57	time::Duration,
58};
59
60/// Maximum blocks per response.
61pub(crate) const MAX_BLOCKS_IN_RESPONSE: usize = 128;
62
63const MAX_BODY_BYTES: usize = 8 * 1024 * 1024;
64const MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER: usize = 2;
65
66mod rep {
67	use sc_network::ReputationChange as Rep;
68
69	/// Reputation change when a peer sent us the same request multiple times.
70	pub const SAME_REQUEST: Rep = Rep::new_fatal("Same block request multiple times");
71
72	/// Reputation change when a peer sent us the same "small" request multiple times.
73	pub const SAME_SMALL_REQUEST: Rep =
74		Rep::new(-(1 << 10), "same small block request multiple times");
75}
76
77/// Generates a `RequestResponseProtocolConfig` for the block request protocol,
78/// refusing incoming requests.
79pub fn generate_protocol_config<
80	Hash: AsRef<[u8]>,
81	B: BlockT,
82	N: NetworkBackend<B, <B as BlockT>::Hash>,
83>(
84	protocol_id: &ProtocolId,
85	genesis_hash: Hash,
86	fork_id: Option<&str>,
87	inbound_queue: async_channel::Sender<IncomingRequest>,
88) -> N::RequestResponseProtocolConfig {
89	N::request_response_config(
90		generate_protocol_name(genesis_hash, fork_id).into(),
91		std::iter::once(generate_legacy_protocol_name(protocol_id).into()).collect(),
92		1024 * 1024,
93		MAX_RESPONSE_SIZE,
94		Duration::from_secs(20),
95		Some(inbound_queue),
96	)
97}
98
99/// Generate the block protocol name from the genesis hash and fork id.
100fn generate_protocol_name<Hash: AsRef<[u8]>>(genesis_hash: Hash, fork_id: Option<&str>) -> String {
101	let genesis_hash = genesis_hash.as_ref();
102	if let Some(fork_id) = fork_id {
103		format!("/{}/{}/sync/2", array_bytes::bytes2hex("", genesis_hash), fork_id)
104	} else {
105		format!("/{}/sync/2", array_bytes::bytes2hex("", genesis_hash))
106	}
107}
108
109/// Generate the legacy block protocol name from chain specific protocol identifier.
110fn generate_legacy_protocol_name(protocol_id: &ProtocolId) -> String {
111	format!("/{}/sync/2", protocol_id.as_ref())
112}
113
114/// The key of [`BlockRequestHandler::seen_requests`].
115#[derive(Eq, PartialEq, Clone)]
116struct SeenRequestsKey<B: BlockT> {
117	peer: PeerId,
118	from: BlockId<B>,
119	max_blocks: usize,
120	direction: Direction,
121	attributes: BlockAttributes,
122	support_multiple_justifications: bool,
123}
124
125#[allow(clippy::derived_hash_with_manual_eq)]
126impl<B: BlockT> Hash for SeenRequestsKey<B> {
127	fn hash<H: Hasher>(&self, state: &mut H) {
128		self.peer.hash(state);
129		self.max_blocks.hash(state);
130		self.direction.hash(state);
131		self.attributes.hash(state);
132		self.support_multiple_justifications.hash(state);
133		match self.from {
134			BlockId::Hash(h) => h.hash(state),
135			BlockId::Number(n) => n.hash(state),
136		}
137	}
138}
139
140/// The value of [`BlockRequestHandler::seen_requests`].
141enum SeenRequestsValue {
142	/// First time we have seen the request.
143	First,
144	/// We have fulfilled the request `n` times.
145	Fulfilled(usize),
146}
147
148/// The full block server implementation of [`BlockServer`]. It handles
149/// the incoming block requests from a remote peer.
150pub struct BlockRequestHandler<B: BlockT, Client> {
151	client: Arc<Client>,
152	request_receiver: async_channel::Receiver<IncomingRequest>,
153	/// Maps from request to number of times we have seen this request.
154	///
155	/// This is used to check if a peer is spamming us with the same request.
156	seen_requests: LruMap<SeenRequestsKey<B>, SeenRequestsValue>,
157}
158
159impl<B, Client> BlockRequestHandler<B, Client>
160where
161	B: BlockT,
162	Client: HeaderBackend<B> + BlockBackend<B> + Send + Sync + 'static,
163{
164	/// Create a new [`BlockRequestHandler`].
165	pub fn new<N: NetworkBackend<B, <B as BlockT>::Hash>>(
166		network: NetworkServiceHandle,
167		protocol_id: &ProtocolId,
168		fork_id: Option<&str>,
169		client: Arc<Client>,
170		num_peer_hint: usize,
171	) -> BlockRelayParams<B, N> {
172		// Reserve enough request slots for one request per peer when we are at the maximum
173		// number of peers.
174		let capacity = std::cmp::max(num_peer_hint, 1);
175		let (tx, request_receiver) = async_channel::bounded(capacity);
176
177		let protocol_config = generate_protocol_config::<_, B, N>(
178			protocol_id,
179			client
180				.block_hash(0u32.into())
181				.ok()
182				.flatten()
183				.expect("Genesis block exists; qed"),
184			fork_id,
185			tx,
186		);
187
188		let capacity = ByLength::new(num_peer_hint.max(1) as u32 * 2);
189		let seen_requests = LruMap::new(capacity);
190
191		BlockRelayParams {
192			server: Box::new(Self { client, request_receiver, seen_requests }),
193			downloader: Arc::new(FullBlockDownloader::new(
194				protocol_config.protocol_name().clone(),
195				network,
196			)),
197			request_response_config: protocol_config,
198		}
199	}
200
201	/// Run [`BlockRequestHandler`].
202	async fn process_requests(&mut self) {
203		while let Some(request) = self.request_receiver.next().await {
204			let IncomingRequest { peer, payload, pending_response } = request;
205
206			match self.handle_request(payload, pending_response, &peer) {
207				Ok(()) => debug!(target: LOG_TARGET, "Handled block request from {}.", peer),
208				Err(e) => debug!(
209					target: LOG_TARGET,
210					"Failed to handle block request from {}: {}", peer, e,
211				),
212			}
213		}
214	}
215
216	fn handle_request(
217		&mut self,
218		payload: Vec<u8>,
219		pending_response: oneshot::Sender<OutgoingResponse>,
220		peer: &PeerId,
221	) -> Result<(), HandleRequestError> {
222		let request = crate::schema::v1::BlockRequest::decode(&payload[..])?;
223
224		let from_block_id = match request.from_block.ok_or(HandleRequestError::MissingFromField)? {
225			FromBlockSchema::Hash(ref h) => {
226				let h = Decode::decode(&mut h.as_ref())?;
227				BlockId::<B>::Hash(h)
228			},
229			FromBlockSchema::Number(ref n) => {
230				let n = Decode::decode(&mut n.as_ref())?;
231				BlockId::<B>::Number(n)
232			},
233		};
234
235		let max_blocks = if request.max_blocks == 0 {
236			MAX_BLOCKS_IN_RESPONSE
237		} else {
238			min(request.max_blocks as usize, MAX_BLOCKS_IN_RESPONSE)
239		};
240
241		let direction =
242			i32::try_into(request.direction).map_err(|_| HandleRequestError::ParseDirection)?;
243
244		let attributes = BlockAttributes::from_be_u32(request.fields)?;
245
246		let support_multiple_justifications = request.support_multiple_justifications;
247
248		let key = SeenRequestsKey {
249			peer: *peer,
250			max_blocks,
251			direction,
252			from: from_block_id,
253			attributes,
254			support_multiple_justifications,
255		};
256
257		let mut reputation_change = None;
258
259		let small_request = attributes
260			.difference(BlockAttributes::HEADER | BlockAttributes::JUSTIFICATION)
261			.is_empty();
262
263		match self.seen_requests.get(&key) {
264			Some(SeenRequestsValue::First) => {},
265			Some(SeenRequestsValue::Fulfilled(ref mut requests)) => {
266				*requests = requests.saturating_add(1);
267
268				if *requests > MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER {
269					reputation_change = Some(if small_request {
270						rep::SAME_SMALL_REQUEST
271					} else {
272						rep::SAME_REQUEST
273					});
274				}
275			},
276			None => {
277				self.seen_requests.insert(key.clone(), SeenRequestsValue::First);
278			},
279		}
280
281		debug!(
282			target: LOG_TARGET,
283			"Handling block request from {peer}: Starting at `{from_block_id:?}` with \
284			maximum blocks of `{max_blocks}`, reputation_change: `{reputation_change:?}`, \
285			small_request `{small_request:?}`, direction `{direction:?}` and \
286			attributes `{attributes:?}`.",
287		);
288
289		let maybe_block_response = if reputation_change.is_none() || small_request {
290			let block_response = self.get_block_response(
291				attributes,
292				from_block_id,
293				direction,
294				max_blocks,
295				support_multiple_justifications,
296			)?;
297
298			// If any of the blocks contains any data, we can consider it as successful request.
299			if block_response
300				.blocks
301				.iter()
302				.any(|b| !b.header.is_empty() || !b.body.is_empty() || b.is_empty_justification)
303			{
304				if let Some(value) = self.seen_requests.get(&key) {
305					// If this is the first time we have processed this request, we need to change
306					// it to `Fulfilled`.
307					if let SeenRequestsValue::First = value {
308						*value = SeenRequestsValue::Fulfilled(1);
309					}
310				}
311			}
312
313			Some(block_response)
314		} else {
315			None
316		};
317
318		debug!(
319			target: LOG_TARGET,
320			"Sending result of block request from {peer} starting at `{from_block_id:?}`: \
321			blocks: {:?}, data: {:?}",
322			maybe_block_response.as_ref().map(|res| res.blocks.len()),
323			maybe_block_response.as_ref().map(|res| res.encoded_len()),
324		);
325
326		let result = if let Some(block_response) = maybe_block_response {
327			let mut data = Vec::with_capacity(block_response.encoded_len());
328			block_response.encode(&mut data)?;
329			Ok(data)
330		} else {
331			Err(())
332		};
333
334		pending_response
335			.send(OutgoingResponse {
336				result,
337				reputation_changes: reputation_change.into_iter().collect(),
338				sent_feedback: None,
339			})
340			.map_err(|_| HandleRequestError::SendResponse)
341	}
342
343	fn get_block_response(
344		&self,
345		attributes: BlockAttributes,
346		mut block_id: BlockId<B>,
347		direction: Direction,
348		max_blocks: usize,
349		support_multiple_justifications: bool,
350	) -> Result<BlockResponse, HandleRequestError> {
351		let get_header = attributes.contains(BlockAttributes::HEADER);
352		let get_body = attributes.contains(BlockAttributes::BODY);
353		let get_indexed_body = attributes.contains(BlockAttributes::INDEXED_BODY);
354		let get_justification = attributes.contains(BlockAttributes::JUSTIFICATION);
355
356		let mut blocks = Vec::new();
357
358		let mut total_size: usize = 0;
359
360		let client_header_from_block_id =
361			|block_id: BlockId<B>| -> Result<Option<B::Header>, HandleRequestError> {
362				if let Some(hash) = self.client.block_hash_from_id(&block_id)? {
363					return self.client.header(hash).map_err(Into::into)
364				}
365				Ok(None)
366			};
367
368		while let Some(header) = client_header_from_block_id(block_id).unwrap_or_default() {
369			let number = *header.number();
370			let hash = header.hash();
371			let parent_hash = *header.parent_hash();
372			let justifications =
373				if get_justification { self.client.justifications(hash)? } else { None };
374
375			let (justifications, justification, is_empty_justification) =
376				if support_multiple_justifications {
377					let justifications = match justifications {
378						Some(v) => v.encode(),
379						None => Vec::new(),
380					};
381					(justifications, Vec::new(), false)
382				} else {
383					// For now we keep compatibility by selecting precisely the GRANDPA one, and not
384					// just the first one. When sending we could have just taken the first one,
385					// since we don't expect there to be any other kind currently, but when
386					// receiving we need to add the engine ID tag.
387					// The ID tag is hardcoded here to avoid depending on the GRANDPA crate, and
388					// will be removed once we remove the backwards compatibility.
389					// See: https://github.com/paritytech/substrate/issues/8172
390					let justification =
391						justifications.and_then(|just| just.into_justification(*b"FRNK"));
392
393					let is_empty_justification =
394						justification.as_ref().map(|j| j.is_empty()).unwrap_or(false);
395
396					let justification = justification.unwrap_or_default();
397
398					(Vec::new(), justification, is_empty_justification)
399				};
400
401			let body = if get_body {
402				match self.client.block_body(hash)? {
403					Some(mut extrinsics) =>
404						extrinsics.iter_mut().map(|extrinsic| extrinsic.encode()).collect(),
405					None => {
406						log::trace!(target: LOG_TARGET, "Missing data for block request.");
407						break
408					},
409				}
410			} else {
411				Vec::new()
412			};
413
414			let indexed_body = if get_indexed_body {
415				match self.client.block_indexed_body(hash)? {
416					Some(transactions) => transactions,
417					None => {
418						log::trace!(
419							target: LOG_TARGET,
420							"Missing indexed block data for block request."
421						);
422						// If the indexed body is missing we still continue returning headers.
423						// Ideally `None` should distinguish a missing body from the empty body,
424						// but the current protobuf based protocol does not allow it.
425						Vec::new()
426					},
427				}
428			} else {
429				Vec::new()
430			};
431
432			let block_data = crate::schema::v1::BlockData {
433				hash: hash.encode(),
434				header: if get_header { header.encode() } else { Vec::new() },
435				body,
436				receipt: Vec::new(),
437				message_queue: Vec::new(),
438				justification,
439				is_empty_justification,
440				justifications,
441				indexed_body,
442			};
443
444			let new_total_size = total_size +
445				block_data.body.iter().map(|ex| ex.len()).sum::<usize>() +
446				block_data.indexed_body.iter().map(|ex| ex.len()).sum::<usize>();
447
448			// Send at least one block, but make sure to not exceed the limit.
449			if !blocks.is_empty() && new_total_size > MAX_BODY_BYTES {
450				break
451			}
452
453			total_size = new_total_size;
454
455			blocks.push(block_data);
456
457			if blocks.len() >= max_blocks as usize {
458				break
459			}
460
461			match direction {
462				Direction::Ascending => block_id = BlockId::Number(number + One::one()),
463				Direction::Descending => {
464					if number.is_zero() {
465						break
466					}
467					block_id = BlockId::Hash(parent_hash)
468				},
469			}
470		}
471
472		Ok(BlockResponse { blocks })
473	}
474}
475
476#[async_trait::async_trait]
477impl<B, Client> BlockServer<B> for BlockRequestHandler<B, Client>
478where
479	B: BlockT,
480	Client: HeaderBackend<B> + BlockBackend<B> + Send + Sync + 'static,
481{
482	async fn run(&mut self) {
483		self.process_requests().await;
484	}
485}
486
487#[derive(Debug, thiserror::Error)]
488enum HandleRequestError {
489	#[error("Failed to decode request: {0}.")]
490	DecodeProto(#[from] prost::DecodeError),
491	#[error("Failed to encode response: {0}.")]
492	EncodeProto(#[from] prost::EncodeError),
493	#[error("Failed to decode block hash: {0}.")]
494	DecodeScale(#[from] codec::Error),
495	#[error("Missing `BlockRequest::from_block` field.")]
496	MissingFromField,
497	#[error("Failed to parse BlockRequest::direction.")]
498	ParseDirection,
499	#[error(transparent)]
500	Client(#[from] sp_blockchain::Error),
501	#[error("Failed to send response.")]
502	SendResponse,
503}
504
505/// The full block downloader implementation of [`BlockDownloader].
506#[derive(Debug)]
507pub struct FullBlockDownloader {
508	protocol_name: ProtocolName,
509	network: NetworkServiceHandle,
510}
511
512impl FullBlockDownloader {
513	fn new(protocol_name: ProtocolName, network: NetworkServiceHandle) -> Self {
514		Self { protocol_name, network }
515	}
516
517	/// Extracts the blocks from the response schema.
518	fn blocks_from_schema<B: BlockT>(
519		&self,
520		request: &BlockRequest<B>,
521		response: BlockResponseSchema,
522	) -> Result<Vec<BlockData<B>>, String> {
523		response
524			.blocks
525			.into_iter()
526			.map(|block_data| {
527				Ok(BlockData::<B> {
528					hash: Decode::decode(&mut block_data.hash.as_ref())?,
529					header: if !block_data.header.is_empty() {
530						Some(Decode::decode(&mut block_data.header.as_ref())?)
531					} else {
532						None
533					},
534					body: if request.fields.contains(BlockAttributes::BODY) {
535						Some(
536							block_data
537								.body
538								.iter()
539								.map(|body| Decode::decode(&mut body.as_ref()))
540								.collect::<Result<Vec<_>, _>>()?,
541						)
542					} else {
543						None
544					},
545					indexed_body: if request.fields.contains(BlockAttributes::INDEXED_BODY) {
546						Some(block_data.indexed_body)
547					} else {
548						None
549					},
550					receipt: if !block_data.receipt.is_empty() {
551						Some(block_data.receipt)
552					} else {
553						None
554					},
555					message_queue: if !block_data.message_queue.is_empty() {
556						Some(block_data.message_queue)
557					} else {
558						None
559					},
560					justification: if !block_data.justification.is_empty() {
561						Some(block_data.justification)
562					} else if block_data.is_empty_justification {
563						Some(Vec::new())
564					} else {
565						None
566					},
567					justifications: if !block_data.justifications.is_empty() {
568						Some(DecodeAll::decode_all(&mut block_data.justifications.as_ref())?)
569					} else {
570						None
571					},
572				})
573			})
574			.collect::<Result<_, _>>()
575			.map_err(|error: codec::Error| error.to_string())
576	}
577}
578
579#[async_trait::async_trait]
580impl<B: BlockT> BlockDownloader<B> for FullBlockDownloader {
581	fn protocol_name(&self) -> &ProtocolName {
582		&self.protocol_name
583	}
584
585	async fn download_blocks(
586		&self,
587		who: PeerId,
588		request: BlockRequest<B>,
589	) -> Result<Result<(Vec<u8>, ProtocolName), RequestFailure>, oneshot::Canceled> {
590		// Build the request protobuf.
591		let bytes = BlockRequestSchema {
592			fields: request.fields.to_be_u32(),
593			from_block: match request.from {
594				FromBlock::Hash(h) => Some(FromBlockSchema::Hash(h.encode())),
595				FromBlock::Number(n) => Some(FromBlockSchema::Number(n.encode())),
596			},
597			direction: request.direction as i32,
598			max_blocks: request.max.unwrap_or(0),
599			support_multiple_justifications: true,
600		}
601		.encode_to_vec();
602
603		let (tx, rx) = oneshot::channel();
604		self.network.start_request(
605			who,
606			self.protocol_name.clone(),
607			bytes,
608			tx,
609			IfDisconnected::ImmediateError,
610		);
611		rx.await
612	}
613
614	fn block_response_into_blocks(
615		&self,
616		request: &BlockRequest<B>,
617		response: Vec<u8>,
618	) -> Result<Vec<BlockData<B>>, BlockResponseError> {
619		// Decode the response protobuf
620		let response_schema = BlockResponseSchema::decode(response.as_slice())
621			.map_err(|error| BlockResponseError::DecodeFailed(error.to_string()))?;
622
623		// Extract the block data from the protobuf
624		self.blocks_from_schema::<B>(request, response_schema)
625			.map_err(|error| BlockResponseError::ExtractionFailed(error.to_string()))
626	}
627}