sc_network_sync/
warp_request_handler.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Substrate.
3
4// Substrate is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8
9// Substrate is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12// GNU General Public License for more details.
13
14// You should have received a copy of the GNU General Public License
15// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
16
17//! Helper for handling (i.e. answering) grandpa warp sync requests from a remote peer.
18
19use codec::Decode;
20use futures::{channel::oneshot, stream::StreamExt};
21use log::debug;
22
23use crate::{
24	strategy::warp::{EncodedProof, WarpProofRequest, WarpSyncProvider},
25	LOG_TARGET,
26};
27use sc_network::{
28	config::ProtocolId,
29	request_responses::{IncomingRequest, OutgoingResponse},
30	NetworkBackend, MAX_RESPONSE_SIZE,
31};
32use sp_runtime::traits::Block as BlockT;
33
34use std::{sync::Arc, time::Duration};
35
36/// Incoming warp requests bounded queue size.
37const MAX_WARP_REQUEST_QUEUE: usize = 20;
38
39/// Generates a `RequestResponseProtocolConfig` for the grandpa warp sync request protocol, refusing
40/// incoming requests.
41pub fn generate_request_response_config<
42	Hash: AsRef<[u8]>,
43	B: BlockT,
44	N: NetworkBackend<B, <B as BlockT>::Hash>,
45>(
46	protocol_id: ProtocolId,
47	genesis_hash: Hash,
48	fork_id: Option<&str>,
49	inbound_queue: async_channel::Sender<IncomingRequest>,
50) -> N::RequestResponseProtocolConfig {
51	N::request_response_config(
52		generate_protocol_name(genesis_hash, fork_id).into(),
53		std::iter::once(generate_legacy_protocol_name(protocol_id).into()).collect(),
54		32,
55		MAX_RESPONSE_SIZE,
56		Duration::from_secs(10),
57		Some(inbound_queue),
58	)
59}
60
61/// Generate the grandpa warp sync protocol name from the genesis hash and fork id.
62fn generate_protocol_name<Hash: AsRef<[u8]>>(genesis_hash: Hash, fork_id: Option<&str>) -> String {
63	let genesis_hash = genesis_hash.as_ref();
64	if let Some(fork_id) = fork_id {
65		format!("/{}/{}/sync/warp", array_bytes::bytes2hex("", genesis_hash), fork_id)
66	} else {
67		format!("/{}/sync/warp", array_bytes::bytes2hex("", genesis_hash))
68	}
69}
70
71/// Generate the legacy grandpa warp sync protocol name from chain specific protocol identifier.
72fn generate_legacy_protocol_name(protocol_id: ProtocolId) -> String {
73	format!("/{}/sync/warp", protocol_id.as_ref())
74}
75
76/// Handler for incoming grandpa warp sync requests from a remote peer.
77pub struct RequestHandler<TBlock: BlockT> {
78	backend: Arc<dyn WarpSyncProvider<TBlock>>,
79	request_receiver: async_channel::Receiver<IncomingRequest>,
80}
81
82impl<TBlock: BlockT> RequestHandler<TBlock> {
83	/// Create a new [`RequestHandler`].
84	pub fn new<Hash: AsRef<[u8]>, N: NetworkBackend<TBlock, <TBlock as BlockT>::Hash>>(
85		protocol_id: ProtocolId,
86		genesis_hash: Hash,
87		fork_id: Option<&str>,
88		backend: Arc<dyn WarpSyncProvider<TBlock>>,
89	) -> (Self, N::RequestResponseProtocolConfig) {
90		let (tx, request_receiver) = async_channel::bounded(MAX_WARP_REQUEST_QUEUE);
91
92		let request_response_config = generate_request_response_config::<_, TBlock, N>(
93			protocol_id,
94			genesis_hash,
95			fork_id,
96			tx,
97		);
98
99		(Self { backend, request_receiver }, request_response_config)
100	}
101
102	fn handle_request(
103		&self,
104		payload: Vec<u8>,
105		pending_response: oneshot::Sender<OutgoingResponse>,
106	) -> Result<(), HandleRequestError> {
107		let request = WarpProofRequest::<TBlock>::decode(&mut &payload[..])?;
108
109		let EncodedProof(proof) = self
110			.backend
111			.generate(request.begin)
112			.map_err(HandleRequestError::InvalidRequest)?;
113
114		pending_response
115			.send(OutgoingResponse {
116				result: Ok(proof),
117				reputation_changes: Vec::new(),
118				sent_feedback: None,
119			})
120			.map_err(|_| HandleRequestError::SendResponse)
121	}
122
123	/// Run [`RequestHandler`].
124	pub async fn run(mut self) {
125		while let Some(request) = self.request_receiver.next().await {
126			let IncomingRequest { peer, payload, pending_response } = request;
127
128			match self.handle_request(payload, pending_response) {
129				Ok(()) => {
130					debug!(target: LOG_TARGET, "Handled grandpa warp sync request from {}.", peer)
131				},
132				Err(e) => debug!(
133					target: LOG_TARGET,
134					"Failed to handle grandpa warp sync request from {}: {}",
135					peer, e,
136				),
137			}
138		}
139	}
140}
141
142#[derive(Debug, thiserror::Error)]
143enum HandleRequestError {
144	#[error("Failed to decode request: {0}.")]
145	DecodeProto(#[from] prost::DecodeError),
146
147	#[error("Failed to encode response: {0}.")]
148	EncodeProto(#[from] prost::EncodeError),
149
150	#[error("Failed to decode block hash: {0}.")]
151	DecodeScale(#[from] codec::Error),
152
153	#[error(transparent)]
154	Client(#[from] sp_blockchain::Error),
155
156	#[error("Invalid request {0}.")]
157	InvalidRequest(#[from] Box<dyn std::error::Error + Send + Sync>),
158
159	#[error("Failed to send response.")]
160	SendResponse,
161}
sc_network_sync/warp_request_handler.rs

sc_network_sync/
warp_request_handler.rs
