Return ResourceUnavailable if we are unable to reconstruct execution payloads (#3365)

## Issue Addressed

Resolves #3351 

## Proposed Changes

Returns a `ResourceUnavailable` rpc error if we are unable to serve full payloads to blocks by root and range requests because the execution layer is not synced.


## Additional Info

This PR also changes the penalties such that a `ResourceUnavailable` error is only penalized if it is an outgoing request. If we are syncing and aren't getting full block responses, then we don't have use for the peer. However, this might not be true for the incoming request case. We let the peer decide in this case if we are still useful or if we should be banned.
cc @divagant-martian please let me know if i'm missing something here.

Author: pawanjay176

beacon_node/lighthouse_network/src/peer_manager/mod.rs

@@ -481,7 +481,15 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
                     // implement a new sync type which tracks these peers and prevents the sync
                     // algorithms from requesting blocks from them (at least for a set period of
                     // time, multiple failures would then lead to a ban).
-                    PeerAction::Fatal
+
+                    match direction {
+                        // If the blocks request was initiated by us, then we have no use of this
+                        // peer and so we ban it.
+                        ConnectionDirection::Outgoing => PeerAction::Fatal,
+                        // If the blocks request was initiated by the peer, then we let the peer decide if
+                        // it wants to continue talking to us, we do not ban the peer.
+                        ConnectionDirection::Incoming => return,
+                    }
                 }
                 RPCResponseErrorCode::ServerError => PeerAction::MidToleranceError,
                 RPCResponseErrorCode::InvalidRequest => PeerAction::LowToleranceError,

beacon_node/network/src/beacon_processor/worker/rpc_methods.rs

@@ -135,6 +135,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         executor.spawn(
             async move {
                 let mut send_block_count = 0;
+                let mut send_response = true;
                 for root in request.block_roots.iter() {
                     match self
                         .chain
@@ -157,6 +158,23 @@ impl<T: BeaconChainTypes> Worker<T> {
                                 "request_root" => ?root
                             );
                         }
+                        Err(BeaconChainError::BlockHashMissingFromExecutionLayer(_)) => {
+                            debug!(
+                                self.log,
+                                "Failed to fetch execution payload for blocks by root request";
+                                "block_root" => ?root,
+                                "reason" => "execution layer not synced",
+                            );
+                            // send the stream terminator
+                            self.send_error_response(
+                                peer_id,
+                                RPCResponseErrorCode::ResourceUnavailable,
+                                "Execution layer not synced".into(),
+                                request_id,
+                            );
+                            send_response = false;
+                            break;
+                        }
                         Err(e) => {
                             debug!(
                                 self.log,
@@ -173,11 +191,13 @@ impl<T: BeaconChainTypes> Worker<T> {
                     "Received BlocksByRoot Request";
                     "peer" => %peer_id,
                     "requested" => request.block_roots.len(),
-                    "returned" => send_block_count
+                    "returned" => %send_block_count
                 );
 
                 // send stream termination
-                self.send_response(peer_id, Response::BlocksByRoot(None), request_id);
+                if send_response {
+                    self.send_response(peer_id, Response::BlocksByRoot(None), request_id);
+                }
                 drop(send_on_drop);
             },
             "load_blocks_by_root_blocks",
@@ -255,6 +275,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         executor.spawn(
             async move {
                 let mut blocks_sent = 0;
+                let mut send_response = true;
 
                 for root in block_roots {
                     match self.chain.get_block(&root).await {
@@ -280,6 +301,23 @@ impl<T: BeaconChainTypes> Worker<T> {
                             );
                             break;
                         }
+                        Err(BeaconChainError::BlockHashMissingFromExecutionLayer(_)) => {
+                            debug!(
+                                self.log,
+                                "Failed to fetch execution payload for blocks by range request";
+                                "block_root" => ?root,
+                                "reason" => "execution layer not synced",
+                            );
+                            // send the stream terminator
+                            self.send_error_response(
+                                peer_id,
+                                RPCResponseErrorCode::ResourceUnavailable,
+                                "Execution layer not synced".into(),
+                                request_id,
+                            );
+                            send_response = false;
+                            break;
+                        }
                         Err(e) => {
                             error!(
                                 self.log,
@@ -320,12 +358,15 @@ impl<T: BeaconChainTypes> Worker<T> {
                     );
                 }
 
-                // send the stream terminator
-                self.send_network_message(NetworkMessage::SendResponse {
-                    peer_id,
-                    response: Response::BlocksByRange(None),
-                    id: request_id,
-                });
+                if send_response {
+                    // send the stream terminator
+                    self.send_network_message(NetworkMessage::SendResponse {
+                        peer_id,
+                        response: Response::BlocksByRange(None),
+                        id: request_id,
+                    });
+                }
+
                 drop(send_on_drop);
             },
             "load_blocks_by_range_blocks",