Move block/state/warpc sync requests/responses to ChainSync

client/network/common/src/sync.rs

@@ -24,14 +24,16 @@ pub mod warp;
 
 use libp2p::PeerId;
 use message::{BlockAnnounce, BlockData, BlockRequest, BlockResponse};
-use sc_consensus::{BlockImportError, BlockImportStatus, IncomingBlock};
+use sc_consensus::{
+ import_queue::RuntimeOrigin, BlockImportError, BlockImportStatus, IncomingBlock,
+};
 use sp_consensus::BlockOrigin;
 use sp_runtime::{
  traits::{Block as BlockT, NumberFor},
  Justifications,
 };
 use std::{any::Any, fmt, fmt::Formatter, task::Poll};
-use warp::{EncodedProof, WarpProofRequest, WarpSyncProgress};
+use warp::WarpSyncProgress;
 
 /// The sync status of a peer we are trying to sync with
 #[derive(Debug)]
@@ -123,7 +125,7 @@ pub enum OnBlockJustification<Block: BlockT> {
  },
 }
 
-/// Result of [`ChainSync::on_state_data`].
+/// Result of `ChainSync::on_state_data`.
 #[derive(Debug)]
 pub enum OnStateData<Block: BlockT> {
  /// The block and state that should be imported.
@@ -132,6 +134,20 @@ pub enum OnStateData<Block: BlockT> {
  Continue,
 }
 
+/// Block or justification request polled from `ChainSync`
+#[derive(Debug)]
+pub enum ImportResult<B: BlockT> {
+ BlockImport(BlockOrigin, Vec<IncomingBlock<B>>),
+ JustificationImport(RuntimeOrigin, B::Hash, NumberFor<B>, Justifications),
+}
+
+/// Value polled from `ChainSync`
+#[derive(Debug)]
+pub enum PollResult<B: BlockT> {
+ Import(ImportResult<B>),
+ Announce(PollBlockAnnounceValidation<B::Header>),
+}
+
 /// Result of [`ChainSync::poll_block_announce_validation`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PollBlockAnnounceValidation<H> {
@@ -186,6 +202,13 @@ pub struct Metrics {
  pub justifications: metrics::Metrics,
 }
 
+#[derive(Debug)]
+pub enum PeerRequest<B: BlockT> {
+ Block(BlockRequest<B>),
+ State,
+ WarpProof,
+}
+
 /// Wrapper for implementation-specific state request.
 ///
 /// NOTE: Implementation must be able to encode and decode it for network purposes.
@@ -250,6 +273,9 @@ pub trait ChainSync<Block: BlockT>: Send {
  /// Returns the current number of peers stored within this state machine.
  fn num_peers(&self) -> usize;
 
+ /// Returns the number of peers we're connected to and that are being queried.
+ fn num_active_peers(&self) -> usize;
+
  /// Handle a new connected peer.
  ///
  /// Call this method whenever we connect to a new peer.
@@ -277,22 +303,6 @@ pub trait ChainSync<Block: BlockT>: Send {
  number: NumberFor<Block>,
  );
 
- /// Get an iterator over all scheduled justification requests.
- fn justification_requests<'a>(
- &'a mut self,
- ) -> Box<dyn Iterator<Item = (PeerId, BlockRequest<Block>)> + 'a>;
-
- /// Get an iterator over all block requests of all peers.
- fn block_requests<'a>(
- &'a mut self,
- ) -> Box<dyn Iterator<Item = (PeerId, BlockRequest<Block>)> + 'a>;
-
- /// Get a state request, if any.
- fn state_request(&mut self) -> Option<(PeerId, OpaqueStateRequest)>;
-
- /// Get a warp sync request, if any.
- fn warp_sync_request(&mut self) -> Option<(PeerId, WarpProofRequest<Block>)>;
-
  /// Handle a response from the remote to a block request that we made.
  ///
  /// `request` must be the original request that triggered `response`.
@@ -307,16 +317,6 @@ pub trait ChainSync<Block: BlockT>: Send {
  response: BlockResponse<Block>,
  ) -> Result<OnBlockData<Block>, BadPeer>;
 
- /// Handle a response from the remote to a state request that we made.
- fn on_state_data(
- &mut self,
- who: &PeerId,
- response: OpaqueStateResponse,
- ) -> Result<OnStateData<Block>, BadPeer>;
-
- /// Handle a response from the remote to a warp proof request that we made.
- fn on_warp_sync_data(&mut self, who: &PeerId, response: EncodedProof) -> Result<(), BadPeer>;
-
  /// Handle a response from the remote to a justification request that we made.
  ///
  /// `request` must be the original request that triggered `response`.
@@ -383,35 +383,20 @@ pub trait ChainSync<Block: BlockT>: Send {
  /// Return some key metrics.
  fn metrics(&self) -> Metrics;
 
- /// Create implementation-specific block request.
- fn create_opaque_block_request(&self, request: &BlockRequest<Block>) -> OpaqueBlockRequest;
-
- /// Encode implementation-specific block request.
- fn encode_block_request(&self, request: &OpaqueBlockRequest) -> Result<Vec<u8>, String>;
-
- /// Decode implementation-specific block response.
- fn decode_block_response(&self, response: &[u8]) -> Result<OpaqueBlockResponse, String>;
-
  /// Access blocks from implementation-specific block response.
  fn block_response_into_blocks(
  &self,
  request: &BlockRequest<Block>,
  response: OpaqueBlockResponse,
  ) -> Result<Vec<BlockData<Block>>, String>;
 
- /// Encode implementation-specific state request.
- fn encode_state_request(&self, request: &OpaqueStateRequest) -> Result<Vec<u8>, String>;
-
- /// Decode implementation-specific state response.
- fn decode_state_response(&self, response: &[u8]) -> Result<OpaqueStateResponse, String>;
-
  /// Advance the state of `ChainSync`
  ///
  /// Internally calls [`ChainSync::poll_block_announce_validation()`] and
  /// this function should be polled until it returns [`Poll::Pending`] to
  /// consume all pending events.
- fn poll(
- &mut self,
-  cx: &mut std::task::Context,
- ) -> Poll<PollBlockAnnounceValidation<Block::Header>>;
+ fn poll(&mut self, cx: &mut std::task::Context) -> Poll<PollResult<Block>>;
+
+ /// Send block request to peer
+ fn send_block_request(&mut self, who: PeerId, request: BlockRequest<Block>);
 }

client/network/src/behaviour.rs

@@ -40,7 +40,6 @@ use sc_network_common::{
  ProtocolName,
  },
  request_responses::{IfDisconnected, ProtocolConfig, RequestFailure},
- sync::{warp::WarpProofRequest, OpaqueBlockRequest, OpaqueStateRequest},
 };
 use sc_peerset::{PeersetHandle, ReputationChange};
 use sp_blockchain::HeaderBackend;
@@ -163,36 +162,6 @@ pub enum BehaviourOut<B: BlockT> {
  messages: Vec<(ProtocolName, Bytes)>,
  },
 
- /// A new block request must be emitted.
- BlockRequest {
- /// Node we send the request to.
- target: PeerId,
- /// Opaque implementation-specific block request.
- request: OpaqueBlockRequest,
- /// One-shot channel to receive the response.
- pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
- },
-
- /// A new state request must be emitted.
- StateRequest {
- /// Node we send the request to.
- target: PeerId,
- /// Opaque implementation-specific state request.
- request: OpaqueStateRequest,
- /// One-shot channel to receive the response.
- pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
- },
-
- /// A new warp sync request must be emitted.
- WarpSyncRequest {
- /// Node we send the request to.
- target: PeerId,
- /// Warp sync request.
- request: WarpProofRequest<B>,
- /// One-shot channel to receive the response.
- pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
- },
-
  /// Now connected to a new peer for syncing purposes.
  SyncConnected(PeerId),
 
@@ -230,21 +199,9 @@ where
  user_agent: String,
  local_public_key: PublicKey,
  disco_config: DiscoveryConfig,
- block_request_protocol_config: ProtocolConfig,
- state_request_protocol_config: ProtocolConfig,
- warp_sync_protocol_config: Option<ProtocolConfig>,
- light_client_request_protocol_config: ProtocolConfig,
- // All remaining request protocol configs.
- mut request_response_protocols: Vec<ProtocolConfig>,
+ request_response_protocols: Vec<ProtocolConfig>,
  peerset: PeersetHandle,
  ) -> Result<Self, request_responses::RegisterError> {
- if let Some(config) = warp_sync_protocol_config {
- request_response_protocols.push(config);
- }
- request_response_protocols.push(block_request_protocol_config);
- request_response_protocols.push(state_request_protocol_config);
- request_response_protocols.push(light_client_request_protocol_config);
-
  Ok(Self {
  substrate,
  peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
@@ -356,12 +313,6 @@ impl<B: BlockT> From<CustomMessageOutcome<B>> for BehaviourOut<B> {
  BehaviourOut::BlockImport(origin, blocks),
  CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) =>
  BehaviourOut::JustificationImport(origin, hash, nb, justification),
- CustomMessageOutcome::BlockRequest { target, request, pending_response } =>
- BehaviourOut::BlockRequest { target, request, pending_response },
- CustomMessageOutcome::StateRequest { target, request, pending_response } =>
- BehaviourOut::StateRequest { target, request, pending_response },
- CustomMessageOutcome::WarpSyncRequest { target, request, pending_response } =>
- BehaviourOut::WarpSyncRequest { target, request, pending_response },
  CustomMessageOutcome::NotificationStreamOpened {
  remote,
  protocol,

client/network/src/config.rs

@@ -101,39 +101,6 @@ where
  /// Block announce protocol configuration
  pub block_announce_config: NonDefaultSetConfig,
 
- /// Request response configuration for the block request protocol.
- ///
- /// [`RequestResponseConfig::name`] is used to tag outgoing block requests with the correct
- /// protocol name. In addition all of [`RequestResponseConfig`] is used to handle incoming
- /// block requests, if enabled.
- ///
- /// Can be constructed either via
- /// `sc_network_sync::block_request_handler::generate_protocol_config` allowing outgoing but
- /// not incoming requests, or constructed via `sc_network_sync::block_request_handler::
- /// BlockRequestHandler::new` allowing both outgoing and incoming requests.
- pub block_request_protocol_config: RequestResponseConfig,
-
- /// Request response configuration for the light client request protocol.
- ///
- /// Can be constructed either via
- /// `sc_network_light::light_client_requests::generate_protocol_config` allowing outgoing but
- /// not incoming requests, or constructed via
- /// `sc_network_light::light_client_requests::handler::LightClientRequestHandler::new`
- /// allowing both outgoing and incoming requests.
- pub light_client_request_protocol_config: RequestResponseConfig,
-
- /// Request response configuration for the state request protocol.
- ///
- /// Can be constructed either via
- /// `sc_network_sync::state_request_handler::generate_protocol_config` allowing outgoing but
- /// not incoming requests, or constructed via
- /// `sc_network_sync::state_request_handler::StateRequestHandler::new` allowing
- /// both outgoing and incoming requests.
- pub state_request_protocol_config: RequestResponseConfig,
-
- /// Optional warp sync protocol config.
- pub warp_sync_protocol_config: Option<RequestResponseConfig>,
-
  /// Request response protocol configurations
  pub request_response_protocol_configs: Vec<RequestResponseConfig>,
 }