Implement request-responses protocols

client/cli/src/params/network_params.rs

@@ -147,6 +147,7 @@ impl NetworkParams {
 			listen_addresses,
 			public_addresses,
 			notifications_protocols: Vec::new(),
+			request_response_protocols: Vec::new(),
 			node_key,
 			node_name: node_name.to_string(),
 			client_version: client_id.to_string(),

client/network/Cargo.toml

@@ -18,6 +18,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 prost-build = "0.6.1"
 
 [dependencies]
+async-trait = "0.1"
 bitflags = "1.2.0"
 bs58 = "0.3.1"
 bytes = "0.5.0"
@@ -65,7 +66,7 @@ zeroize = "1.0.0"
 [dependencies.libp2p]
 version = "0.21.1"
 default-features = false
-features = ["identify", "kad", "mdns", "mplex", "noise", "ping", "tcp-async-std", "websocket", "yamux"]
+features = ["identify", "kad", "mdns", "mplex", "noise", "ping", "request-response", "tcp-async-std", "websocket", "yamux"]
 
 [dev-dependencies]
 async-std = "1.6.2"

client/network/src/behaviour.rs

@@ -16,9 +16,9 @@
 
 use crate::{
 	config::{ProtocolId, Role}, block_requests, light_client_handler, finality_requests,
-	peer_info, discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
+	request_responses, peer_info, discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
 	protocol::{message::{self, Roles}, CustomMessageOutcome, Protocol},
-	Event, ObservedRole, DhtEvent, ExHashT,
+	service::maybe_utf8_bytes_to_string, Event, ObservedRole, DhtEvent, ExHashT,
 };
 
 use codec::Encode as _;
@@ -37,6 +37,10 @@ use std::{
 	time::Duration,
 };
 
+pub use crate::request_responses::{
+	InboundError, InboundFailure, OutboundFailure, RequestId, SendRequestError
+};
+
 /// General behaviour of the network. Combines all protocols together.
 #[derive(NetworkBehaviour)]
 #[behaviour(out_event = "BehaviourOut<B>", poll_method = "poll")]
@@ -48,6 +52,8 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	peer_info: peer_info::PeerInfoBehaviour,
 	/// Discovers nodes of the network.
 	discovery: DiscoveryBehaviour,
+	/// Generic request-reponse protocols.
+	request_responses: request_responses::RequestResponsesBehaviour,
 	/// Block request handling.
 	block_requests: block_requests::BlockRequests<B>,
 	/// Finality proof request handling.
@@ -74,22 +80,40 @@ pub enum BehaviourOut<B: BlockT> {
 	RandomKademliaStarted(ProtocolId),
 
 	/// We have received a request from a peer and answered it.
-	AnsweredRequest {
+	///
+	/// This event is generated for statistics purposes.
+	InboundRequest {
 		/// Peer which sent us a request.
 		peer: PeerId,
 		/// Protocol name of the request.
-		protocol: Vec<u8>,
-		/// Time it took to build the response.
-		build_time: Duration,
+		protocol: Cow<'static, str>,
+		/// If `Ok`, contains the time elapsed between when we received the request and when we
+		/// sent back the response. If `Err`, the error that happened.
+		outcome: Result<Duration, InboundError>,
 	},
+
+	/// A request initiated using [`Behaviour::send_request`] has succeeded or failed.
+	RequestFinished {
+		/// Request that has succeeded.
+		request_id: RequestId,
+		/// Response sent by the remote or reason for failure.
+		outcome: Result<Vec<u8>, OutboundFailure>,
+	},
+
 	/// Started a new request with the given node.
-	RequestStarted {
+	///
+	/// This event is for statistics purposes only. The request and response handling are entirely
+	/// internal to the behaviour.
+	OpaqueRequestStarted {
 		peer: PeerId,
 		/// Protocol name of the request.
 		protocol: Vec<u8>,
 	},
 	/// Finished, successfully or not, a previously-started request.
-	RequestFinished {
+	///
+	/// This event is for statistics purposes only. The request and response handling are entirely
+	/// internal to the behaviour.
+	OpaqueRequestFinished {
 		/// Who we were requesting.
 		peer: PeerId,
 		/// Protocol name of the request.
@@ -116,17 +140,20 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 		finality_proof_requests: finality_requests::FinalityProofRequests<B>,
 		light_client_handler: light_client_handler::LightClientHandler<B>,
 		disco_config: DiscoveryConfig,
-	) -> Self {
-		Behaviour {
+		request_response_protocols: Vec<request_responses::ProtocolConfig>,
+	) -> Result<Self, request_responses::RegisterError> {
+		Ok(Behaviour {
 			substrate,
 			peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
 			discovery: disco_config.finish(),
+			request_responses:
+				request_responses::RequestResponsesBehaviour::new(request_response_protocols.into_iter())?,
 			block_requests,
 			finality_proof_requests,
 			light_client_handler,
 			events: VecDeque::new(),
 			role,
-		}
+		})
 	}
 
 	/// Returns the list of nodes that we know exist in the network.
@@ -163,6 +190,16 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 		self.peer_info.node(peer_id)
 	}
 
+	/// Initiates sending a request.
+	///
+	/// An error is returned if we are not connected to the target peer of if the protocol doesn't
+	/// match one that has been registered.
+	pub fn send_request(&mut self, target: &PeerId, protocol: &str, request: Vec<u8>)
+		-> Result<RequestId, SendRequestError>
+	{
+		self.request_responses.send_request(target, protocol, request)
+	}
+
 	/// Registers a new notifications protocol.
 	///
 	/// After that, you can call `write_notifications`.
@@ -255,18 +292,18 @@ Behaviour<B, H> {
 			CustomMessageOutcome::BlockRequest { target, request } => {
 				match self.block_requests.send_request(&target, request) {
 					block_requests::SendRequestOutcome::Ok => {
-						self.events.push_back(BehaviourOut::RequestStarted {
+						self.events.push_back(BehaviourOut::OpaqueRequestStarted {
 							peer: target,
 							protocol: self.block_requests.protocol_name().to_vec(),
 						});
 					},
 					block_requests::SendRequestOutcome::Replaced { request_duration, .. } => {
-						self.events.push_back(BehaviourOut::RequestFinished {
+						self.events.push_back(BehaviourOut::OpaqueRequestFinished {
 							peer: target.clone(),
 							protocol: self.block_requests.protocol_name().to_vec(),
 							request_duration,
 						});
-						self.events.push_back(BehaviourOut::RequestStarted {
+						self.events.push_back(BehaviourOut::OpaqueRequestStarted {
 							peer: target,
 							protocol: self.block_requests.protocol_name().to_vec(),
 						});
@@ -307,18 +344,41 @@ Behaviour<B, H> {
 	}
 }
 
+impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<request_responses::Event> for Behaviour<B, H> {
+	fn inject_event(&mut self, event: request_responses::Event) {
+		match event {
+			request_responses::Event::InboundRequest { peer, protocol, outcome } => {
+				self.events.push_back(BehaviourOut::InboundRequest {
+					peer,
+					protocol,
+					outcome,
+				});
+			}
+
+			request_responses::Event::OutboundFinished { request_id, outcome } => {
+				self.events.push_back(BehaviourOut::RequestFinished {
+					request_id,
+					outcome,
+				});
+			},
+		}
+	}
+}
+
 impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<block_requests::Event<B>> for Behaviour<B, H> {
 	fn inject_event(&mut self, event: block_requests::Event<B>) {
 		match event {
 			block_requests::Event::AnsweredRequest { peer, total_handling_time } => {
-				self.events.push_back(BehaviourOut::AnsweredRequest {
+				let protocol = maybe_utf8_bytes_to_string(&self.block_requests.protocol_name())
+					.into_owned();
+				self.events.push_back(BehaviourOut::InboundRequest {
 					peer,
-					protocol: self.block_requests.protocol_name().to_vec(),
-					build_time: total_handling_time,
+					protocol: From::from(protocol),
+					outcome: Ok(total_handling_time),
 				});
 			},
 			block_requests::Event::Response { peer, original_request: _, response, request_duration } => {
-				self.events.push_back(BehaviourOut::RequestFinished {
+				self.events.push_back(BehaviourOut::OpaqueRequestFinished {
 					peer: peer.clone(),
 					protocol: self.block_requests.protocol_name().to_vec(),
 					request_duration,
@@ -330,7 +390,7 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<block_requests::Event<B
 			block_requests::Event::RequestTimeout { peer, request_duration, .. } => {
 				// There doesn't exist any mechanism to report cancellations or timeouts yet, so
 				// we process them by disconnecting the node.
-				self.events.push_back(BehaviourOut::RequestFinished {
+				self.events.push_back(BehaviourOut::OpaqueRequestFinished {
 					peer: peer.clone(),
 					protocol: self.block_requests.protocol_name().to_vec(),
 					request_duration,

client/network/src/config.rs

@@ -23,6 +23,7 @@
 
 pub use crate::chain::{Client, FinalityProofProvider};
 pub use crate::on_demand_layer::{AlwaysBadChecker, OnDemand};
+pub use crate::request_responses::{IncomingRequest, ProtocolConfig as RequestResponseConfig};
 pub use libp2p::{identity, core::PublicKey, wasm_ext::ExtTransport, build_multiaddr};
 
 // Note: this re-export shouldn't be part of the public API of the crate and will be removed in
@@ -34,9 +35,10 @@ use crate::ExHashT;
 
 use core::{fmt, iter};
 use futures::future;
-use libp2p::identity::{ed25519, Keypair};
-use libp2p::wasm_ext;
-use libp2p::{multiaddr, Multiaddr, PeerId};
+use libp2p::{
+	identity::{ed25519, Keypair},
+	multiaddr, wasm_ext, Multiaddr, PeerId,
+};
 use prometheus_endpoint::Registry;
 use sp_consensus::{block_validation::BlockAnnounceValidator, import_queue::ImportQueue};
 use sp_runtime::{traits::Block as BlockT, ConsensusEngineId};
@@ -407,6 +409,8 @@ pub struct NetworkConfiguration {
 	/// List of notifications protocols that the node supports. Must also include a
 	/// `ConsensusEngineId` for backwards-compatibility.
 	pub notifications_protocols: Vec<(ConsensusEngineId, Cow<'static, [u8]>)>,
+	/// List of request-response protocols that the node supports.
+	pub request_response_protocols: Vec<RequestResponseConfig>,
 	/// Maximum allowed number of incoming connections.
 	pub in_peers: u32,
 	/// Number of outgoing connections we're trying to maintain.
@@ -442,6 +446,7 @@ impl NetworkConfiguration {
 			boot_nodes: Vec::new(),
 			node_key,
 			notifications_protocols: Vec::new(),
+			request_response_protocols: Vec::new(),
 			in_peers: 25,
 			out_peers: 75,
 			reserved_nodes: Vec::new(),
@@ -458,9 +463,7 @@ impl NetworkConfiguration {
 			allow_non_globals_in_dht: false,
 		}
 	}
-}
 
-impl NetworkConfiguration {
 	/// Create new default configuration for localhost-only connection with random port (useful for testing)
 	pub fn new_local() -> NetworkConfiguration {
 		let mut config = NetworkConfiguration::new(

client/network/src/error.rs

@@ -21,7 +21,7 @@
 use crate::config::TransportConfig;
 use libp2p::{PeerId, Multiaddr};
 
-use std::fmt;
+use std::{borrow::Cow, fmt};
 
 /// Result type alias for the network.
 pub type Result<T> = std::result::Result<T, Error>;
@@ -61,6 +61,12 @@ pub enum Error {
 		/// The invalid addresses.
 		addresses: Vec<Multiaddr>,
 	},
+	/// The same request-response protocol has been registered multiple times.
+	#[display(fmt = "Request-response protocol registered multiple times: {}", protocol)]
+	DuplicateRequestResponseProtocol {
+		/// Name of the protocol registered multiple times.
+		protocol: Cow<'static, str>,
+	},
 }
 
 // Make `Debug` use the `Display` implementation.
@@ -78,6 +84,7 @@ impl std::error::Error for Error {
 			Error::DuplicateBootnode { .. } => None,
 			Error::Prometheus(ref err) => Some(err),
 			Error::AddressesForAnotherTransport { .. } => None,
+			Error::DuplicateRequestResponseProtocol { .. } => None,
 		}
 	}
 }

client/network/src/lib.rs

@@ -251,6 +251,7 @@ mod finality_requests;
 mod light_client_handler;
 mod on_demand_layer;
 mod protocol;
+mod request_responses;
 mod schema;
 mod service;
 mod transport;
@@ -260,13 +261,10 @@ pub mod config;
 pub mod error;
 pub mod network_state;
 
-pub use service::{NetworkService, NetworkWorker};
-pub use protocol::PeerInfo;
-pub use protocol::event::{Event, DhtEvent, ObservedRole};
-pub use protocol::sync::SyncState;
-pub use libp2p::{Multiaddr, PeerId};
 #[doc(inline)]
-pub use libp2p::multiaddr;
+pub use libp2p::{multiaddr, Multiaddr, PeerId};
+pub use protocol::{event::{DhtEvent, Event, ObservedRole}, sync::SyncState, PeerInfo};
+pub use service::{NetworkService, NetworkWorker, OutboundFailure};
 
 pub use sc_peerset::ReputationChange;
 use sp_runtime::traits::{Block as BlockT, NumberFor};