Modular block request handler

cumulus/client/service/src/lib.rs

@@ -484,6 +484,7 @@ where
 		import_queue,
 		block_announce_validator_builder: Some(Box::new(move |_| block_announce_validator)),
 		warp_sync_params,
+		block_relay: None,
 	})
 }
 

polkadot/node/service/src/lib.rs

@@ -894,6 +894,7 @@ pub fn new_full<OverseerGenerator: OverseerGen>(
 			import_queue,
 			block_announce_validator_builder: None,
 			warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
+			block_relay: None,
 		})?;
 
 	if config.offchain_worker.enabled {

substrate/bin/node-template/node/src/service.rs

@@ -183,6 +183,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 			import_queue,
 			block_announce_validator_builder: None,
 			warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
+			block_relay: None,
 		})?;
 
 	if config.offchain_worker.enabled {

substrate/bin/node/cli/src/service.rs

@@ -388,6 +388,7 @@ pub fn new_full_base(
 			import_queue,
 			block_announce_validator_builder: None,
 			warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
+			block_relay: None,
 		})?;
 
 	let role = config.role.clone();

substrate/client/network/common/src/sync.rs

@@ -22,12 +22,12 @@ pub mod message;
 pub mod metrics;
 pub mod warp;
 
-use crate::{role::Roles, sync::message::BlockAnnounce, types::ReputationChange};
+use crate::{role::Roles, types::ReputationChange};
 use futures::Stream;
 
 use libp2p_identity::PeerId;
 
-use message::{BlockData, BlockRequest, BlockResponse};
+use message::{BlockAnnounce, BlockRequest, BlockResponse};
 use sc_consensus::{import_queue::RuntimeOrigin, IncomingBlock};
 use sp_consensus::BlockOrigin;
 use sp_runtime::{
@@ -226,28 +226,6 @@ impl fmt::Debug for OpaqueStateResponse {
 	}
 }
 
-/// Wrapper for implementation-specific block request.
-///
-/// NOTE: Implementation must be able to encode and decode it for network purposes.
-pub struct OpaqueBlockRequest(pub Box<dyn Any + Send>);
-
-impl fmt::Debug for OpaqueBlockRequest {
-	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-		f.debug_struct("OpaqueBlockRequest").finish()
-	}
-}
-
-/// Wrapper for implementation-specific block response.
-///
-/// NOTE: Implementation must be able to encode and decode it for network purposes.
-pub struct OpaqueBlockResponse(pub Box<dyn Any + Send>);
-
-impl fmt::Debug for OpaqueBlockResponse {
-	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-		f.debug_struct("OpaqueBlockResponse").finish()
-	}
-}
-
 /// Provides high-level status of syncing.
 #[async_trait::async_trait]
 pub trait SyncStatusProvider<Block: BlockT>: Send + Sync {
@@ -392,13 +370,6 @@ pub trait ChainSync<Block: BlockT>: Send {
 	/// Return some key metrics.
 	fn metrics(&self) -> Metrics;
 
-	/// Access blocks from implementation-specific block response.
-	fn block_response_into_blocks(
-		&self,
-		request: &BlockRequest<Block>,
-		response: OpaqueBlockResponse,
-	) -> Result<Vec<BlockData<Block>>, String>;
-
 	/// Advance the state of `ChainSync`
 	fn poll(&mut self, cx: &mut std::task::Context) -> Poll<()>;
 

substrate/client/network/sync/src/block_relay_protocol.rs

@@ -0,0 +1,72 @@
+// Copyright Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Block relay protocol related definitions.
+
+use futures::channel::oneshot;
+use libp2p::PeerId;
+use sc_network::request_responses::{ProtocolConfig, RequestFailure};
+use sc_network_common::sync::message::{BlockData, BlockRequest};
+use sp_runtime::traits::Block as BlockT;
+use std::sync::Arc;
+
+/// The serving side of the block relay protocol. It runs a single instance
+/// of the server task that processes the incoming protocol messages.
+#[async_trait::async_trait]
+pub trait BlockServer<Block: BlockT>: Send {
+	/// Starts the protocol processing.
+	async fn run(&mut self);
+}
+
+/// The client side stub to download blocks from peers. This is a handle
+/// that can be used to initiate concurrent downloads.
+#[async_trait::async_trait]
+pub trait BlockDownloader<Block: BlockT>: Send + Sync {
+	/// Performs the protocol specific sequence to fetch the blocks from the peer.
+	/// Output: if the download succeeds, the response is a `Vec<u8>` which is
+	/// in a format specific to the protocol implementation. The block data
+	/// can be extracted from this response using [`BlockDownloader::block_response_into_blocks`].
+	async fn download_blocks(
+		&self,
+		who: PeerId,
+		request: BlockRequest<Block>,
+	) -> Result<Result<Vec<u8>, RequestFailure>, oneshot::Canceled>;
+
+	/// Parses the protocol specific response to retrieve the block data.
+	fn block_response_into_blocks(
+		&self,
+		request: &BlockRequest<Block>,
+		response: Vec<u8>,
+	) -> Result<Vec<BlockData<Block>>, BlockResponseError>;
+}
+
+/// Errors returned by [`BlockDownloader::block_response_into_blocks`].
+#[derive(Debug)]
+pub enum BlockResponseError {
+	/// Failed to decode the response bytes.
+	DecodeFailed(String),
+
+	/// Failed to extract the blocks from the decoded bytes.
+	ExtractionFailed(String),
+}
+
+/// Block relay specific params for network creation, specified in
+/// ['sc_service::BuildNetworkParams'].
+pub struct BlockRelayParams<Block: BlockT> {
+	pub server: Box<dyn BlockServer<Block>>,
+	pub downloader: Arc<dyn BlockDownloader<Block>>,
+	pub request_response_config: ProtocolConfig,
+}