Avoid processing redundant RPC blocks (sigp#4179)

## Proposed Changes

We already make some attempts to avoid processing RPC blocks when a block from the same proposer is already being processed through gossip. This PR strengthens that guarantee by using the existing cache for `observed_block_producers` to inform whether an RPC block's processing should be delayed.

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -348,7 +348,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// in recent epochs.
     pub(crate) observed_sync_aggregators: RwLock<ObservedSyncAggregators<T::EthSpec>>,
     /// Maintains a record of which validators have proposed blocks for each slot.
-    pub(crate) observed_block_producers: RwLock<ObservedBlockProducers<T::EthSpec>>,
+    pub observed_block_producers: RwLock<ObservedBlockProducers<T::EthSpec>>,
     /// Maintains a record of which validators have submitted voluntary exits.
     pub(crate) observed_voluntary_exits: Mutex<ObservedOperations<SignedVoluntaryExit, T::EthSpec>>,
     /// Maintains a record of which validators we've seen proposer slashings for.

beacon_node/network/src/beacon_processor/work_reprocessing_queue.rs

@@ -55,7 +55,7 @@ pub const QUEUED_ATTESTATION_DELAY: Duration = Duration::from_secs(12);
 pub const QUEUED_LIGHT_CLIENT_UPDATE_DELAY: Duration = Duration::from_secs(12);
 
 /// For how long to queue rpc blocks before sending them back for reprocessing.
-pub const QUEUED_RPC_BLOCK_DELAY: Duration = Duration::from_secs(3);
+pub const QUEUED_RPC_BLOCK_DELAY: Duration = Duration::from_secs(4);
 
 /// Set an arbitrary upper-bound on the number of queued blocks to avoid DoS attacks. The fact that
 /// we signature-verify blocks before putting them in the queue *should* protect against this, but
@@ -520,7 +520,7 @@ impl<T: BeaconChainTypes> ReprocessQueue<T> {
                     return;
                 }
 
-                // Queue the block for 1/4th of a slot
+                // Queue the block for 1/3rd of a slot
                 self.rpc_block_delay_queue
                     .insert(rpc_block, QUEUED_RPC_BLOCK_DELAY);
             }

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

@@ -83,6 +83,52 @@ impl<T: BeaconChainTypes> Worker<T> {
                 return;
             }
         };
+        // Check if a block from this proposer is already known. If so, defer processing until later
+        // to avoid wasting time processing duplicates.
+        let proposal_already_known = self
+            .chain
+            .observed_block_producers
+            .read()
+            .proposer_has_been_observed(block.message())
+            .map_err(|e| {
+                error!(
+                    self.log,
+                    "Failed to check observed proposers";
+                    "error" => ?e,
+                    "source" => "rpc",
+                    "block_root" => %block_root
+                );
+            })
+            .unwrap_or(true);
+        if proposal_already_known {
+            debug!(
+                self.log,
+                "Delaying processing of duplicate RPC block";
+                "block_root" => ?block_root,
+                "proposer" => block.message().proposer_index(),
+                "slot" => block.slot()
+            );
+
+            // Send message to work reprocess queue to retry the block
+            let reprocess_msg = ReprocessQueueMessage::RpcBlock(QueuedRpcBlock {
+                block_root,
+                block: block.clone(),
+                process_type,
+                seen_timestamp,
+                should_process: true,
+            });
+
+            if reprocess_tx.try_send(reprocess_msg).is_err() {
+                error!(
+                    self.log,
+                    "Failed to inform block import";
+                    "source" => "rpc",
+                    "block_root" => %block_root
+                );
+            }
+            return;
+        }
+
         let slot = block.slot();
         let parent_root = block.message().parent_root();
         let result = self