Disconnect sync peer on rpc HandlerReject

Author: pawanjay176

beacon_node/eth2_libp2p/src/behaviour/mod.rs

@@ -77,6 +77,8 @@ pub enum BehaviourEvent<TSpec: EthSpec> {
         id: RequestId,
         /// The peer to which this request was sent.
         peer_id: PeerId,
+        /// The rpc error that was received.
+        error: RPCError,
     },
     RequestReceived {
         /// The peer that sent the request.
@@ -891,7 +893,7 @@ impl<TSpec: EthSpec> NetworkBehaviourEventProcess<RPCMessage<TSpec>> for Behavio
                         );
                         // inform failures of requests comming outside the behaviour
                         if !matches!(id, RequestId::Behaviour) {
-                            self.add_event(BehaviourEvent::RPCFailed { peer_id, id });
+                            self.add_event(BehaviourEvent::RPCFailed { peer_id, id, error });
                         }
                     }
                 }

beacon_node/network/src/router/mod.rs

@@ -11,8 +11,8 @@ use crate::error;
 use crate::service::NetworkMessage;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::{
-    rpc::RequestId, MessageId, NetworkGlobals, PeerId, PeerRequestId, PubsubMessage, Request,
-    Response,
+    rpc::{RPCError, RequestId},
+    MessageId, NetworkGlobals, PeerId, PeerRequestId, PubsubMessage, Request, Response,
 };
 use futures::prelude::*;
 use processor::Processor;
@@ -59,6 +59,7 @@ pub enum RouterMessage<T: EthSpec> {
     RPCFailed {
         peer_id: PeerId,
         request_id: RequestId,
+        error: RPCError,
     },
     /// A gossip message has been received. The fields are: message id, the peer that sent us this
     /// message, the message itself and a bool which indicates if the message should be processed
@@ -141,8 +142,9 @@ impl<T: BeaconChainTypes> Router<T> {
             RouterMessage::RPCFailed {
                 peer_id,
                 request_id,
+                error,
             } => {
-                self.processor.on_rpc_error(peer_id, request_id);
+                self.processor.on_rpc_error(peer_id, request_id, error);
             }
             RouterMessage::PubsubMessage(id, peer_id, gossip, should_process) => {
                 self.handle_gossip(id, peer_id, gossip, should_process);

beacon_node/network/src/router/processor.rs

@@ -95,10 +95,10 @@ impl<T: BeaconChainTypes> Processor<T> {
 
     /// An error occurred during an RPC request. The state is maintained by the sync manager, so
     /// this function notifies the sync manager of the error.
-    pub fn on_rpc_error(&mut self, peer_id: PeerId, request_id: RequestId) {
+    pub fn on_rpc_error(&mut self, peer_id: PeerId, request_id: RequestId, error: RPCError) {
         // Check if the failed RPC belongs to sync
         if let RequestId::Sync(id) = request_id {
-            self.send_to_sync(SyncMessage::RPCError(peer_id, id));
+            self.send_to_sync(SyncMessage::RPCError(peer_id, id, error));
         }
     }
 

beacon_node/network/src/service.rs

@@ -613,10 +613,10 @@ fn spawn_service<T: BeaconChainTypes>(
                                     });
 
                             }
-                            BehaviourEvent::RPCFailed{id, peer_id} => {
+                            BehaviourEvent::RPCFailed{id, peer_id, error} => {
                                 let _ = service
                                     .router_send
-                                    .send(RouterMessage::RPCFailed{ peer_id, request_id: id})
+                                    .send(RouterMessage::RPCFailed{ peer_id, request_id: id, error})
                                     .map_err(|_| {
                                         debug!(service.log, "Failed to send RPC to router");
                                     });

beacon_node/network/src/sync/manager.rs

@@ -41,6 +41,7 @@ use crate::beacon_processor::{ProcessId, WorkEvent as BeaconWorkEvent};
 use crate::service::NetworkMessage;
 use crate::status::ToStatusMessage;
 use beacon_chain::{BeaconChain, BeaconChainTypes, BlockError};
+use eth2_libp2p::rpc::RPCError;
 use eth2_libp2p::rpc::{methods::MAX_REQUEST_BLOCKS, BlocksByRootRequest, GoodbyeReason};
 use eth2_libp2p::types::{NetworkGlobals, SyncState};
 use eth2_libp2p::SyncInfo;
@@ -102,7 +103,7 @@ pub enum SyncMessage<T: EthSpec> {
     Disconnect(PeerId),
 
     /// An RPC Error has occurred on a request.
-    RPCError(PeerId, RequestId),
+    RPCError(PeerId, RequestId, RPCError),
 
     /// A batch has been processed by the block processor thread.
     BatchProcessed {
@@ -576,7 +577,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
         }
     }
 
-    fn inject_error(&mut self, peer_id: PeerId, request_id: RequestId) {
+    fn inject_error(&mut self, peer_id: PeerId, request_id: RequestId, error: RPCError) {
         trace!(self.log, "Sync manager received a failed RPC");
         // remove any single block lookups
         if self.single_block_lookups.remove(&request_id).is_some() {
@@ -597,6 +598,12 @@ impl<T: BeaconChainTypes> SyncManager<T> {
             return;
         }
 
+        // `HandlerRejected` error signifies that the peer has disconnected from us.
+        // Remove the peer from any syncing chain mappings.
+        if let RPCError::HandlerRejected = error {
+            self.peer_disconnect(&peer_id);
+        }
+
         // otherwise, this is a range sync issue, notify the range sync
         self.range_sync
             .inject_error(&mut self.network, peer_id, request_id);
@@ -912,8 +919,8 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                     SyncMessage::Disconnect(peer_id) => {
                         self.peer_disconnect(&peer_id);
                     }
-                    SyncMessage::RPCError(peer_id, request_id) => {
-                        self.inject_error(peer_id, request_id);
+                    SyncMessage::RPCError(peer_id, request_id, error) => {
+                        self.inject_error(peer_id, request_id, error);
                     }
                     SyncMessage::BatchProcessed {
                         chain_id,