sc_network_sync/
warp_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) grandpa warp sync requests from a remote peer.
19
20use codec::Decode;
21use futures::{channel::oneshot, stream::StreamExt};
22use log::debug;
23
24use crate::{
25	strategy::warp::{EncodedProof, WarpProofRequest, WarpSyncProvider},
26	LOG_TARGET,
27};
28use sc_network::{
29	config::ProtocolId,
30	request_responses::{IncomingRequest, OutgoingResponse},
31	NetworkBackend, MAX_RESPONSE_SIZE,
32};
33use sp_runtime::traits::Block as BlockT;
34
35use std::{sync::Arc, time::Duration};
36
37/// Incoming warp requests bounded queue size.
38const MAX_WARP_REQUEST_QUEUE: usize = 20;
39
40/// Generates a `RequestResponseProtocolConfig` for the grandpa warp sync request protocol, refusing
41/// incoming requests.
42pub fn generate_request_response_config<
43	Hash: AsRef<[u8]>,
44	B: BlockT,
45	N: NetworkBackend<B, <B as BlockT>::Hash>,
46>(
47	protocol_id: ProtocolId,
48	genesis_hash: Hash,
49	fork_id: Option<&str>,
50	inbound_queue: async_channel::Sender<IncomingRequest>,
51) -> N::RequestResponseProtocolConfig {
52	N::request_response_config(
53		generate_protocol_name(genesis_hash, fork_id).into(),
54		std::iter::once(generate_legacy_protocol_name(protocol_id).into()).collect(),
55		32,
56		MAX_RESPONSE_SIZE,
57		Duration::from_secs(10),
58		Some(inbound_queue),
59	)
60}
61
62/// Generate the grandpa warp sync protocol name from the genesis hash and fork id.
63fn generate_protocol_name<Hash: AsRef<[u8]>>(genesis_hash: Hash, fork_id: Option<&str>) -> String {
64	let genesis_hash = genesis_hash.as_ref();
65	if let Some(fork_id) = fork_id {
66		format!("/{}/{}/sync/warp", array_bytes::bytes2hex("", genesis_hash), fork_id)
67	} else {
68		format!("/{}/sync/warp", array_bytes::bytes2hex("", genesis_hash))
69	}
70}
71
72/// Generate the legacy grandpa warp sync protocol name from chain specific protocol identifier.
73fn generate_legacy_protocol_name(protocol_id: ProtocolId) -> String {
74	format!("/{}/sync/warp", protocol_id.as_ref())
75}
76
77/// Handler for incoming grandpa warp sync requests from a remote peer.
78pub struct RequestHandler<TBlock: BlockT> {
79	backend: Arc<dyn WarpSyncProvider<TBlock>>,
80	request_receiver: async_channel::Receiver<IncomingRequest>,
81}
82
83impl<TBlock: BlockT> RequestHandler<TBlock> {
84	/// Create a new [`RequestHandler`].
85	pub fn new<Hash: AsRef<[u8]>, N: NetworkBackend<TBlock, <TBlock as BlockT>::Hash>>(
86		protocol_id: ProtocolId,
87		genesis_hash: Hash,
88		fork_id: Option<&str>,
89		backend: Arc<dyn WarpSyncProvider<TBlock>>,
90	) -> (Self, N::RequestResponseProtocolConfig) {
91		let (tx, request_receiver) = async_channel::bounded(MAX_WARP_REQUEST_QUEUE);
92
93		let request_response_config = generate_request_response_config::<_, TBlock, N>(
94			protocol_id,
95			genesis_hash,
96			fork_id,
97			tx,
98		);
99
100		(Self { backend, request_receiver }, request_response_config)
101	}
102
103	fn handle_request(
104		&self,
105		payload: Vec<u8>,
106		pending_response: oneshot::Sender<OutgoingResponse>,
107	) -> Result<(), HandleRequestError> {
108		let request = WarpProofRequest::<TBlock>::decode(&mut &payload[..])?;
109
110		let EncodedProof(proof) = self
111			.backend
112			.generate(request.begin)
113			.map_err(HandleRequestError::InvalidRequest)?;
114
115		pending_response
116			.send(OutgoingResponse {
117				result: Ok(proof),
118				reputation_changes: Vec::new(),
119				sent_feedback: None,
120			})
121			.map_err(|_| HandleRequestError::SendResponse)
122	}
123
124	/// Run [`RequestHandler`].
125	pub async fn run(mut self) {
126		while let Some(request) = self.request_receiver.next().await {
127			let IncomingRequest { peer, payload, pending_response } = request;
128
129			match self.handle_request(payload, pending_response) {
130				Ok(()) => {
131					debug!(target: LOG_TARGET, "Handled grandpa warp sync request from {}.", peer)
132				},
133				Err(e) => debug!(
134					target: LOG_TARGET,
135					"Failed to handle grandpa warp sync request from {}: {}",
136					peer, e,
137				),
138			}
139		}
140	}
141}
142
143#[derive(Debug, thiserror::Error)]
144enum HandleRequestError {
145	#[error("Failed to decode request: {0}.")]
146	DecodeProto(#[from] prost::DecodeError),
147
148	#[error("Failed to encode response: {0}.")]
149	EncodeProto(#[from] prost::EncodeError),
150
151	#[error("Failed to decode block hash: {0}.")]
152	DecodeScale(#[from] codec::Error),
153
154	#[error(transparent)]
155	Client(#[from] sp_blockchain::Error),
156
157	#[error("Invalid request {0}.")]
158	InvalidRequest(#[from] Box<dyn std::error::Error + Send + Sync>),
159
160	#[error("Failed to send response.")]
161	SendResponse,
162}
sc_network_sync/warp_request_handler.rs

sc_network_sync/
warp_request_handler.rs
