client/network: Use request response for block requests (#7478)

* client/network: Add scaffolding for finality req to use req resp
	#sc

* client/network/src/finality_requests: Remove

* client/network/src/behaviour: Pass request id down to sync

* client/network: Use request response for block requests

* client/network: Move handler logic into *_*_handler.rs

* client/network: Track ongoing finality requests in protocol.rs

* client/network: Remove commented out finalization initialization

* client/network: Add docs for request handlers

* client/network/finality_request_handler: Log errors

* client/network/block_request_handler: Log errors

* client/network: Format

* client/network: Handle block request failure

* protocols/network: Fix tests

* client/network/src/behaviour: Handle request sending errors

* client/network: Move response handling into custom method

* client/network/protocol: Handle block response errors

* client/network/protocol: Remove tracking of obsolete requests

* client/network/protocol: Remove block request start time tracking

This will be handled generically via request-responses.

* client/network/protocol: Refactor on_*_request_started

* client/network: Pass protocol config instead of protocol name

* client/network: Pass protocol config in tests

* client/network/config: Document request response configs

* client/network/src/_request_handler: Document protocol config gen

* client/network/src/protocol: Document Peer request values

* client/network: Rework request response to always use oneshot

* client/network: Unified metric reporting for all request protocols

* client/network: Move protobuf parsing into protocol.rs

* client/network/src/protocol: Return pending events after poll

* client/network: Improve error handling and documentation

* client/network/behaviour: Remove outdated error types

* Update client/network/src/block_request_handler.rs

Co-authored-by: Ashley <[email protected]>

* Update client/network/src/finality_request_handler.rs

Co-authored-by: Ashley <[email protected]>

* client/network/protocol: Reduce reputation on timeout

* client/network/protocol: Refine reputation changes

* client/network/block_request_handler: Set and explain queue length

* client/service: Deny block requests when light client

* client/service: Fix role matching

* client: Enforce line width

* client/network/request_responses: Fix unit tests

* client/network: Expose time to build response via metrics

* client/network/request_responses: Fix early connection closed error

* client/network/protocol: Fix line length

* client/network/protocol: Disconnect on most request failures

* client/network/protocol: Disconnect peer when oneshot is canceled

* client/network/protocol: Disconnect peer even when connection closed

* client/network/protocol: Remove debugging log line

* client/network/request_response: Use Clone::clone for error

* client/network/request_response: Remove outdated comment

With libp2p v0.33.0 libp2p-request-response properly sends inbound
failures on connections being closed.

Co-authored-by: Addie Wagenknecht <[email protected]>
Co-authored-by: Ashley <[email protected]>

Author: 3 people

.maintain/sentry-node/docker-compose.yml

@@ -47,9 +47,9 @@ services:
       - "--validator"
       - "--alice"
       - "--sentry-nodes"
-      - "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
+      - "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
       - "--reserved-nodes"
-      - "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
+      - "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
       # Not only bind to localhost.
       - "--unsafe-ws-external"
       - "--unsafe-rpc-external"
@@ -83,11 +83,11 @@ services:
       - "--port"
       - "30333"
       - "--sentry"
-      - "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
+      - "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
       - "--reserved-nodes"
-      - "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
+      - "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
       - "--bootnodes"
-      - "/dns/validator-b/tcp/30333/p2p/QmSVnNf9HwVMT1Y4cK1P6aoJcEZjmoTXpjKBmAABLMnZEk"
+      - "/dns/validator-b/tcp/30333/p2p/12D3KooWHdiAxVd8uMQR1hGWXccidmfCwLqcMpGwR6QcTP6QRMuD"
       - "--no-telemetry"
       - "--rpc-cors"
       - "all"
@@ -118,9 +118,9 @@ services:
       - "--validator"
       - "--bob"
       - "--bootnodes"
-      - "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
+      - "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
       - "--bootnodes"
-      - "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
+      - "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
       - "--no-telemetry"
       - "--rpc-cors"
       - "all"

Cargo.lock

@@ -17,20 +17,22 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
-	config::{ProtocolId, Role}, block_requests, light_client_handler,
-	peer_info, request_responses, discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
+	config::{ProtocolId, Role}, light_client_handler, peer_info, request_responses,
+	discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
 	protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
 	ObservedRole, DhtEvent, ExHashT,
 };
 
 use bytes::Bytes;
 use codec::Encode as _;
+use futures::channel::oneshot;
 use libp2p::NetworkBehaviour;
 use libp2p::core::{Multiaddr, PeerId, PublicKey};
 use libp2p::identify::IdentifyInfo;
 use libp2p::kad::record;
 use libp2p::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters};
 use log::debug;
+use prost::Message;
 use sp_consensus::{BlockOrigin, import_queue::{IncomingBlock, Origin}};
 use sp_runtime::{traits::{Block as BlockT, NumberFor}, Justification};
 use std::{
@@ -42,7 +44,7 @@ use std::{
 };
 
 pub use crate::request_responses::{
-	ResponseFailure, InboundFailure, RequestFailure, OutboundFailure, RequestId, SendRequestError
+	ResponseFailure, InboundFailure, RequestFailure, OutboundFailure, RequestId,
 };
 
 /// General behaviour of the network. Combines all protocols together.
@@ -58,8 +60,6 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	discovery: DiscoveryBehaviour,
 	/// Generic request-reponse protocols.
 	request_responses: request_responses::RequestResponsesBehaviour,
-	/// Block request handling.
-	block_requests: block_requests::BlockRequests<B>,
 	/// Light client request handling.
 	light_client_handler: light_client_handler::LightClientHandler<B>,
 
@@ -70,6 +70,11 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	/// Role of our local node, as originally passed from the configuration.
 	#[behaviour(ignore)]
 	role: Role,
+
+	/// Protocol name used to send out block requests via
+	/// [`request_responses::RequestResponsesBehaviour`].
+	#[behaviour(ignore)]
+	block_request_protocol_name: String,
 }
 
 /// Event generated by `Behaviour`.
@@ -93,34 +98,18 @@ pub enum BehaviourOut<B: BlockT> {
 		result: Result<Duration, ResponseFailure>,
 	},
 
-	/// 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.
-		result: Result<Vec<u8>, RequestFailure>,
-	},
-
-	/// Started a new request with the given node.
-	///
-	/// 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: String,
-	},
-	/// Finished, successfully or not, a previously-started request.
+	/// A request has succeeded or failed.
 	///
-	/// This event is for statistics purposes only. The request and response handling are entirely
-	/// internal to the behaviour.
-	OpaqueRequestFinished {
-		/// Who we were requesting.
+	/// This event is generated for statistics purposes.
+	RequestFinished {
+		/// Peer that we send a request to.
 		peer: PeerId,
-		/// Protocol name of the request.
-		protocol: String,
-		/// How long before the response came or the request got cancelled.
-		request_duration: Duration,
+		/// Name of the protocol in question.
+		protocol: Cow<'static, str>,
+		/// Duration the request took.
+		duration: Duration,
+		/// Result of the request.
+		result: Result<(), RequestFailure>,
 	},
 
 	/// Opened a substream with the given node with the given notifications protocol.
@@ -180,21 +169,28 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 		role: Role,
 		user_agent: String,
 		local_public_key: PublicKey,
-		block_requests: block_requests::BlockRequests<B>,
 		light_client_handler: light_client_handler::LightClientHandler<B>,
 		disco_config: DiscoveryConfig,
-		request_response_protocols: Vec<request_responses::ProtocolConfig>,
+		// Block request protocol config.
+		block_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);
+
 		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,
 			light_client_handler,
 			events: VecDeque::new(),
 			role,
+
+			block_request_protocol_name,
 		})
 	}
 
@@ -236,13 +232,14 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	}
 
 	/// 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)
+	pub fn send_request(
+		&mut self,
+		target: &PeerId,
+		protocol: &str,
+		request: Vec<u8>,
+		pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
+	) {
+		self.request_responses.send_request(target, protocol, request, pending_response)
 	}
 
 	/// Registers a new notifications protocol.
@@ -331,28 +328,20 @@ Behaviour<B, H> {
 				self.events.push_back(BehaviourOut::BlockImport(origin, blocks)),
 			CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) =>
 				self.events.push_back(BehaviourOut::JustificationImport(origin, hash, nb, justification)),
-			CustomMessageOutcome::BlockRequest { target, request } => {
-				match self.block_requests.send_request(&target, request) {
-					block_requests::SendRequestOutcome::Ok => {
-						self.events.push_back(BehaviourOut::OpaqueRequestStarted {
-							peer: target,
-							protocol: self.block_requests.protocol_name().to_owned(),
-						});
-					},
-					block_requests::SendRequestOutcome::Replaced { request_duration, .. } => {
-						self.events.push_back(BehaviourOut::OpaqueRequestFinished {
-							peer: target.clone(),
-							protocol: self.block_requests.protocol_name().to_owned(),
-							request_duration,
-						});
-						self.events.push_back(BehaviourOut::OpaqueRequestStarted {
-							peer: target,
-							protocol: self.block_requests.protocol_name().to_owned(),
-						});
-					}
-					block_requests::SendRequestOutcome::NotConnected |
-					block_requests::SendRequestOutcome::EncodeError(_) => {},
+			CustomMessageOutcome::BlockRequest { target, request, pending_response } => {
+				let mut buf = Vec::with_capacity(request.encoded_len());
+				if let Err(err) = request.encode(&mut buf) {
+					log::warn!(
+						target: "sync",
+						"Failed to encode block request {:?}: {:?}",
+						request, err
+					);
+					return
 				}
+
+				self.request_responses.send_request(
+					&target, &self.block_request_protocol_name, buf, pending_response,
+				);
 			},
 			CustomMessageOutcome::NotificationStreamOpened { remote, protocols, roles, notifications_sink } => {
 				let role = reported_roles_to_observed_role(&self.role, &remote, roles);
@@ -401,51 +390,15 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<request_responses::Even
 					result,
 				});
 			}
-
-			request_responses::Event::RequestFinished { request_id, result } => {
+			request_responses::Event::RequestFinished { peer, protocol, duration, result } => {
 				self.events.push_back(BehaviourOut::RequestFinished {
-					request_id,
-					result,
+					peer, protocol, duration, result,
 				});
 			},
 		}
 	}
 }
 
-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::InboundRequest {
-					peer,
-					protocol: self.block_requests.protocol_name().to_owned().into(),
-					result: Ok(total_handling_time),
-				});
-			},
-			block_requests::Event::Response { peer, response, request_duration } => {
-				self.events.push_back(BehaviourOut::OpaqueRequestFinished {
-					peer: peer.clone(),
-					protocol: self.block_requests.protocol_name().to_owned(),
-					request_duration,
-				});
-				let ev = self.substrate.on_block_response(peer, response);
-				self.inject_event(ev);
-			}
-			block_requests::Event::RequestCancelled { peer, request_duration, .. } |
-			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::OpaqueRequestFinished {
-					peer: peer.clone(),
-					protocol: self.block_requests.protocol_name().to_owned(),
-					request_duration,
-				});
-				self.substrate.on_block_request_failed(&peer);
-			}
-		}
-	}
-}
-
 impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<peer_info::PeerInfoEvent>
 	for Behaviour<B, H> {
 	fn inject_event(&mut self, event: peer_info::PeerInfoEvent) {