1use 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
60pub(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 pub const SAME_REQUEST: Rep = Rep::new_fatal("Same block request multiple times");
71
72 pub const SAME_SMALL_REQUEST: Rep =
74 Rep::new(-(1 << 10), "same small block request multiple times");
75}
76
77pub 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
99fn 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
109fn generate_legacy_protocol_name(protocol_id: &ProtocolId) -> String {
111 format!("/{}/sync/2", protocol_id.as_ref())
112}
113
114#[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
140enum SeenRequestsValue {
142 First,
144 Fulfilled(usize),
146}
147
148pub struct BlockRequestHandler<B: BlockT, Client> {
151 client: Arc<Client>,
152 request_receiver: async_channel::Receiver<IncomingRequest>,
153 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 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 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 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 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 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 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 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 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#[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 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 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 let response_schema = BlockResponseSchema::decode(response.as_slice())
621 .map_err(|error| BlockResponseError::DecodeFailed(error.to_string()))?;
622
623 self.blocks_from_schema::<B>(request, response_schema)
625 .map_err(|error| BlockResponseError::ExtractionFailed(error.to_string()))
626 }
627}