client/network: Use request response for light client requests

client/network/src/behaviour.rs

@@ -17,14 +17,15 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
-	config::{ProtocolId, Role}, light_client_handler, peer_info, request_responses,
+	config::{ProtocolId, Role},
 	discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
 	protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
+	peer_info, request_responses, light_client_requests,
 	ObservedRole, DhtEvent, ExHashT,
 };
 
 use bytes::Bytes;
-use futures::channel::oneshot;
+use futures::{channel::oneshot, stream::StreamExt};
 use libp2p::NetworkBehaviour;
 use libp2p::core::{Multiaddr, PeerId, PublicKey};
 use libp2p::identify::IdentifyInfo;
@@ -59,8 +60,6 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	discovery: DiscoveryBehaviour,
 	/// Generic request-reponse protocols.
 	request_responses: request_responses::RequestResponsesBehaviour,
-	/// Light client request handling.
-	light_client_handler: light_client_handler::LightClientHandler<B>,
 
 	/// Queue of events to produce for the outside.
 	#[behaviour(ignore)]
@@ -70,6 +69,10 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	#[behaviour(ignore)]
 	role: Role,
 
+	/// Light client request handling.
+	#[behaviour(ignore)]
+	light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
+
 	/// Protocol name used to send out block requests via
 	/// [`request_responses::RequestResponsesBehaviour`].
 	#[behaviour(ignore)]
@@ -174,24 +177,26 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 		role: Role,
 		user_agent: String,
 		local_public_key: PublicKey,
-		light_client_handler: light_client_handler::LightClientHandler<B>,
+		light_client_request_sender: light_client_requests::sender::LightClientRequestSender<B>,
 		disco_config: DiscoveryConfig,
-		// Block request protocol config.
 		block_request_protocol_config: request_responses::ProtocolConfig,
+		light_client_request_protocol_config: request_responses::ProtocolConfig,
 		// All remaining request protocol configs.
 		mut request_response_protocols: Vec<request_responses::ProtocolConfig>,
 	) -> Result<Self, request_responses::RegisterError> {
 		// Extract protocol name and add to `request_response_protocols`.
 		let block_request_protocol_name = block_request_protocol_config.name.to_string();
 		request_response_protocols.push(block_request_protocol_config);
 
+		request_response_protocols.push(light_client_request_protocol_config);
+
 		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())?,
-			light_client_handler,
+			light_client_request_sender,
 			events: VecDeque::new(),
 			role,
 
@@ -268,8 +273,11 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	}
 
 	/// Issue a light client request.
-	pub fn light_client_request(&mut self, r: light_client_handler::Request<B>) -> Result<(), light_client_handler::Error> {
-		self.light_client_handler.request(r)
+	pub fn light_client_request(
+		&mut self,
+		r: light_client_requests::sender::Request<B>,
+	) -> Result<(), light_client_requests::sender::SendRequestError> {
+		self.light_client_request_sender.request(r)
 	}
 }
 
@@ -289,13 +297,6 @@ fn reported_roles_to_observed_role(local_role: &Role, remote: &PeerId, roles: Ro
 	}
 }
 
-impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<void::Void> for
-Behaviour<B, H> {
-	fn inject_event(&mut self, event: void::Void) {
-		void::unreachable(event)
-	}
-}
-
 impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<CustomMessageOutcome<B>> for
 Behaviour<B, H> {
 	fn inject_event(&mut self, event: CustomMessageOutcome<B>) {
@@ -343,12 +344,18 @@ Behaviour<B, H> {
 				self.events.push_back(BehaviourOut::NotificationsReceived { remote, messages });
 			},
 			CustomMessageOutcome::PeerNewBest(peer_id, number) => {
-				self.light_client_handler.update_best_block(&peer_id, number);
+				self.light_client_request_sender.update_best_block(&peer_id, number);
+			}
+			CustomMessageOutcome::SyncConnected(peer_id) => {
+				// TODO: Should this be tied to sync connections, or to connections in general?
+				self.light_client_request_sender.inject_connected(peer_id);
+				self.events.push_back(BehaviourOut::SyncConnected(peer_id))
+			}
+			CustomMessageOutcome::SyncDisconnected(peer_id) => {
+				// TODO: Should this be tied to sync connections, or to connections in general?
+				self.light_client_request_sender.inject_disconnected(peer_id);
+				self.events.push_back(BehaviourOut::SyncDisconnected(peer_id))
 			}
-			CustomMessageOutcome::SyncConnected(peer_id) =>
-				self.events.push_back(BehaviourOut::SyncConnected(peer_id)),
-			CustomMessageOutcome::SyncDisconnected(peer_id) =>
-				self.events.push_back(BehaviourOut::SyncDisconnected(peer_id)),
 			CustomMessageOutcome::None => {}
 		}
 	}
@@ -443,7 +450,20 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 }
 
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
-	fn poll<TEv>(&mut self, _: &mut Context, _: &mut impl PollParameters) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
+	fn poll<TEv>(
+		&mut self,
+		cx: &mut Context,
+		_: &mut impl PollParameters,
+	) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
+		use light_client_requests::sender::OutEvent;
+		while let Poll::Ready(Some(event)) = self.light_client_request_sender.poll_next_unpin(cx) {
+			match event {
+				OutEvent::SendRequest { target, request, pending_response, protocol_name } => {
+					self.request_responses.send_request(&target, &protocol_name, request, pending_response)
+				}
+			}
+		}
+
 		if let Some(event) = self.events.pop_front() {
 			return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event))
 		}

client/network/src/block_request_handler.rs

@@ -39,7 +39,7 @@ const MAX_BLOCKS_IN_RESPONSE: usize = 128;
 const MAX_BODY_BYTES: usize = 8 * 1024 * 1024;
 
 /// Generates a [`ProtocolConfig`] for the block request protocol, refusing incoming requests.
-pub fn generate_protocol_config(protocol_id: ProtocolId) -> ProtocolConfig {
+pub fn generate_protocol_config(protocol_id: &ProtocolId) -> ProtocolConfig {
 	ProtocolConfig {
 		name: generate_protocol_name(protocol_id).into(),
 		max_request_size: 1024 * 1024,
@@ -50,7 +50,10 @@ pub fn generate_protocol_config(protocol_id: ProtocolId) -> ProtocolConfig {
 }
 
 /// Generate the block protocol name from chain specific protocol identifier.
-fn generate_protocol_name(protocol_id: ProtocolId) -> String {
+//
+// Visibility `pub(crate)` to allow `crate::light_client_requests::sender` to generate block request
+// protocol name and send block requests.
+pub(crate) fn generate_protocol_name(protocol_id: &ProtocolId) -> String {
 	let mut s = String::new();
 	s.push_str("/");
 	s.push_str(protocol_id.as_ref());
@@ -66,7 +69,7 @@ pub struct BlockRequestHandler<B> {
 
 impl <B: BlockT> BlockRequestHandler<B> {
 	/// Create a new [`BlockRequestHandler`].
-	pub fn new(protocol_id: ProtocolId, client: Arc<dyn Client<B>>) -> (Self, ProtocolConfig) {
+	pub fn new(protocol_id: &ProtocolId, client: Arc<dyn Client<B>>) -> (Self, ProtocolConfig) {
 		// Rate of arrival multiplied with the waiting time in the queue equals the queue length.
 		//
 		// An average Polkadot sentry node serves less than 5 requests per second. The 95th percentile
@@ -82,6 +85,22 @@ impl <B: BlockT> BlockRequestHandler<B> {
 		(Self { client, request_receiver }, protocol_config)
 	}
 
+	/// Run [`BlockRequestHandler`].
+	pub async fn run(mut self) {
+		while let Some(request) = self.request_receiver.next().await {
+			let IncomingRequest { peer, payload, pending_response } = request;
+
+			match self.handle_request(payload, pending_response) {
+				Ok(()) => debug!(target: LOG_TARGET, "Handled block request from {}.", peer),
+				Err(e) => debug!(
+					target: LOG_TARGET,
+					"Failed to handle block request from {}: {}",
+					peer, e,
+				),
+			}
+		}
+	}
+
 	fn handle_request(
 		&self,
 		payload: Vec<u8>,
@@ -186,22 +205,6 @@ impl <B: BlockT> BlockRequestHandler<B> {
 			reputation_changes: Vec::new(),
 		}).map_err(|_| HandleRequestError::SendResponse)
 	}
-
-	/// Run [`BlockRequestHandler`].
-	pub async fn run(mut self) {
-		while let Some(request) = self.request_receiver.next().await {
-			let IncomingRequest { peer, payload, pending_response } = request;
-
-			match self.handle_request(payload, pending_response) {
-				Ok(()) => debug!(target: LOG_TARGET, "Handled block request from {}.", peer),
-				Err(e) => debug!(
-					target: LOG_TARGET,
-					"Failed to handle block request from {}: {}",
-					peer, e,
-				),
-			}
-		}
-	}
 }
 
 #[derive(derive_more::Display, derive_more::From)]

client/network/src/config.rs

@@ -111,6 +111,14 @@ pub struct Params<B: BlockT, H: ExHashT> {
 	/// [`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 [`light_client_requests::generate_protocol_config`] allowing
+	/// outgoing but not incoming requests, or constructed via
+	/// [`light_client_requests::handler::LightClientRequestHandler::new`] allowing both outgoing
+	/// and incoming requests.
+	pub light_client_request_protocol_config: RequestResponseConfig,
 }
 
 /// Role of the local node.

client/network/src/gossip/tests.rs

@@ -17,6 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::block_request_handler::BlockRequestHandler;
+use crate::light_client_requests::handler::LightClientRequestHandler;
 use crate::gossip::QueuedSender;
 use crate::{config,  Event, NetworkService, NetworkWorker};
 
@@ -96,7 +97,16 @@ fn build_test_full_node(network_config: config::NetworkConfiguration)
 
 	let block_request_protocol_config = {
 		let (handler, protocol_config) = BlockRequestHandler::new(
-			protocol_id.clone(),
+			&protocol_id,
+			client.clone(),
+		);
+		async_std::task::spawn(handler.run().boxed());
+		protocol_config
+	};
+
+	let light_client_request_protocol_config = {
+		let (handler, protocol_config) = LightClientRequestHandler::new(
+			&protocol_id,
 			client.clone(),
 		);
 		async_std::task::spawn(handler.run().boxed());
@@ -117,6 +127,7 @@ fn build_test_full_node(network_config: config::NetworkConfiguration)
 		),
 		metrics_registry: None,
 		block_request_protocol_config,
+		light_client_request_protocol_config,
 	})
 	.unwrap();
 

client/network/src/lib.rs

@@ -249,7 +249,6 @@ mod behaviour;
 mod chain;
 mod peer_info;
 mod discovery;
-mod light_client_handler;
 mod on_demand_layer;
 mod protocol;
 mod request_responses;
@@ -259,6 +258,7 @@ mod transport;
 mod utils;
 
 pub mod block_request_handler;
+pub mod light_client_requests;
 pub mod config;
 pub mod error;
 pub mod gossip;