Prevent double import of blocks (#2647)

## Issue Addressed

Resolves #2611 

## Proposed Changes

Adds a duplicate block root cache to the `BeaconProcessor`. Adds the block root to the cache before calling `process_gossip_block` and `process_rpc_block`. Since `process_rpc_block` is called only for single block lookups, we don't have to worry about batched block imports.

The block is imported from the source(gossip/rpc) that arrives first. The block that arrives second is not imported to avoid the db access issue.
There are 2 cases:
1. Block that arrives second is from rpc: In this case, we return an optimistic `BlockError::BlockIsAlreadyKnown` to sync.
2. Block that arrives second is from gossip: In this case, we only do gossip verification and forwarding but don't import the block into the the beacon chain.

## Additional info
Splits up `process_gossip_block` function to `process_gossip_unverified_block` and `process_gossip_verified_block`.

beacon_node/network/src/beacon_processor/mod.rs

@@ -39,6 +39,7 @@
 //! task.
 
 use crate::{metrics, service::NetworkMessage, sync::SyncMessage};
+use beacon_chain::parking_lot::Mutex;
 use beacon_chain::{BeaconChain, BeaconChainTypes, BlockError, GossipVerifiedBlock};
 use futures::stream::{Stream, StreamExt};
 use futures::task::Poll;
@@ -47,13 +48,13 @@ use lighthouse_network::{
     Client, MessageId, NetworkGlobals, PeerId, PeerRequestId,
 };
 use slog::{crit, debug, error, trace, warn, Logger};
-use std::cmp;
 use std::collections::VecDeque;
 use std::fmt;
 use std::pin::Pin;
 use std::sync::{Arc, Weak};
 use std::task::Context;
 use std::time::{Duration, Instant};
+use std::{cmp, collections::HashSet};
 use task_executor::TaskExecutor;
 use tokio::sync::{mpsc, oneshot};
 use types::{
@@ -284,6 +285,55 @@ impl<T> LifoQueue<T> {
     }
 }
 
+/// A handle that sends a message on the provided channel to a receiver when it gets dropped.
+///
+/// The receiver task is responsible for removing the provided `entry` from the `DuplicateCache`
+/// and perform any other necessary cleanup.
+pub struct DuplicateCacheHandle {
+    entry: Hash256,
+    cache: DuplicateCache,
+}
+
+impl Drop for DuplicateCacheHandle {
+    fn drop(&mut self) {
+        self.cache.remove(&self.entry);
+    }
+}
+
+/// A simple  cache for detecting duplicate block roots across multiple threads.
+#[derive(Clone, Default)]
+pub struct DuplicateCache {
+    inner: Arc<Mutex<HashSet<Hash256>>>,
+}
+
+impl DuplicateCache {
+    /// Checks if the given block_root exists and inserts it into the cache if
+    /// it doesn't exist.
+    ///
+    /// Returns a `Some(DuplicateCacheHandle)` if the block_root was successfully
+    /// inserted and `None` if the block root already existed in the cache.
+    ///
+    /// The handle removes the entry from the cache when it is dropped. This ensures that any unclean
+    /// shutdowns in the worker tasks does not leave inconsistent state in the cache.
+    pub fn check_and_insert(&self, block_root: Hash256) -> Option<DuplicateCacheHandle> {
+        let mut inner = self.inner.lock();
+        if inner.insert(block_root) {
+            Some(DuplicateCacheHandle {
+                entry: block_root,
+                cache: self.clone(),
+            })
+        } else {
+            None
+        }
+    }
+
+    /// Remove the given block_root from the cache.
+    pub fn remove(&self, block_root: &Hash256) {
+        let mut inner = self.inner.lock();
+        inner.remove(block_root);
+    }
+}
+
 /// An event to be processed by the manager task.
 pub struct WorkEvent<T: BeaconChainTypes> {
     drop_during_sync: bool,
@@ -787,6 +837,7 @@ pub struct BeaconProcessor<T: BeaconChainTypes> {
     pub executor: TaskExecutor,
     pub max_workers: usize,
     pub current_workers: usize,
+    pub importing_blocks: DuplicateCache,
     pub log: Logger,
 }
 
@@ -1302,6 +1353,8 @@ impl<T: BeaconChainTypes> BeaconProcessor<T> {
             log: self.log.clone(),
         };
 
+        let duplicate_cache = self.importing_blocks.clone();
+
         trace!(
             self.log,
             "Spawning beacon processor worker";
@@ -1373,7 +1426,8 @@ impl<T: BeaconChainTypes> BeaconProcessor<T> {
                         peer_id,
                         peer_client,
                         *block,
-                        work_reprocessing_tx,
+                        work_reprocessing_tx.clone(),
+                        duplicate_cache,
                         seen_timestamp,
                     ),
                     /*
@@ -1455,7 +1509,12 @@ impl<T: BeaconChainTypes> BeaconProcessor<T> {
                      * Verification for beacon blocks received during syncing via RPC.
                      */
                     Work::RpcBlock { block, result_tx } => {
-                        worker.process_rpc_block(*block, result_tx, work_reprocessing_tx)
+                        worker.process_rpc_block(
+                            *block,
+                            result_tx,
+                            work_reprocessing_tx.clone(),
+                            duplicate_cache,
+                        );
                     }
                     /*
                      * Verification for a chain segment (multiple blocks).