Add early check for RPC block relevancy (#2289)

## Issue Addressed

NA

## Proposed Changes

When observing `jemallocator` heap profiles and Grafana, it became clear that Lighthouse is spending significant RAM/CPU on processing blocks from the RPC. On investigation, it seems that we are loading the parent of the block *before* we check to see if the block is already known. This is a big waste of resources.

This PR adds an additional `check_block_relevancy` call as the first thing we do when we try to process a `SignedBeaconBlock` via the RPC (or other similar methods). Ultimately, `check_block_relevancy` will be called again later in the block processing flow. It's a very light function and I don't think trying to optimize it out is worth the risk of a bad block slipping through. 

Also adds a `New RPC block received` info log when we process a new RPC block. This seems like interesting and infrequent info.

## Additional Info

NA

Author: paulhauner

beacon_node/beacon_chain/src/block_verification.rs

@@ -690,15 +690,14 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
     /// Returns an error if the block is invalid, or if the block was unable to be verified.
     pub fn new(
         block: SignedBeaconBlock<T::EthSpec>,
+        block_root: Hash256,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
         let (mut parent, block) = load_parent(block, chain)?;
 
         // Reject any block that exceeds our limit on skipped slots.
         check_block_skip_slots(chain, parent.beacon_block.slot(), &block.message)?;
 
-        let block_root = get_block_root(&block);
-
         let state = cheap_state_advance_to_obtain_committees(
             &mut parent.pre_state,
             parent.beacon_state_root,
@@ -726,10 +725,11 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
     /// As for `new` above but producing `BlockSlashInfo`.
     pub fn check_slashable(
         block: SignedBeaconBlock<T::EthSpec>,
+        block_root: Hash256,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockSlashInfo<BlockError<T::EthSpec>>> {
         let header = block.signed_block_header();
-        Self::new(block, chain).map_err(|e| BlockSlashInfo::from_early_error(header, e))
+        Self::new(block, block_root, chain).map_err(|e| BlockSlashInfo::from_early_error(header, e))
     }
 
     /// Finishes signature verification on the provided `GossipVerifedBlock`. Does not re-verify
@@ -814,7 +814,11 @@ impl<T: BeaconChainTypes> IntoFullyVerifiedBlock<T> for SignedBeaconBlock<T::Eth
         self,
         chain: &BeaconChain<T>,
     ) -> Result<FullyVerifiedBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
-        SignatureVerifiedBlock::check_slashable(self, chain)?
+        // Perform an early check to prevent wasting time on irrelevant blocks.
+        let block_root = check_block_relevancy(&self, None, chain)
+            .map_err(|e| BlockSlashInfo::SignatureNotChecked(self.signed_block_header(), e))?;
+
+        SignatureVerifiedBlock::check_slashable(self, block_root, chain)?
             .into_fully_verified_block_slashable(chain)
     }
 

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

@@ -6,7 +6,7 @@ use crate::sync::manager::SyncMessage;
 use crate::sync::{BatchProcessResult, ChainId};
 use beacon_chain::{BeaconChainTypes, BlockError, ChainSegmentResult};
 use eth2_libp2p::PeerId;
-use slog::{crit, debug, error, trace, warn};
+use slog::{crit, debug, error, info, trace, warn};
 use types::{Epoch, Hash256, SignedBeaconBlock};
 
 /// Id associated to a block processing request, either a batch or a single block.
@@ -28,10 +28,20 @@ impl<T: BeaconChainTypes> Worker<T> {
         block: SignedBeaconBlock<T::EthSpec>,
         result_tx: BlockResultSender<T::EthSpec>,
     ) {
+        let slot = block.slot();
         let block_result = self.chain.process_block(block);
 
         metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
 
+        if let Ok(root) = &block_result {
+            info!(
+                self.log,
+                "New RPC block received";
+                "slot" => slot,
+                "hash" => %root
+            );
+        }
+
         if result_tx.send(block_result).is_err() {
             crit!(self.log, "Failed return sync block result");
         }