Merge branch 'eip4844' into deneb-free-blobs

beacon_node/network/src/beacon_processor/tests.rs

@@ -185,6 +185,7 @@ impl TestRig {
             None,
             meta_data,
             vec![],
+            false,
             &log,
         ));
 

beacon_node/network/src/beacon_processor/work_reprocessing_queue.rs

@@ -56,7 +56,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
@@ -521,7 +521,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

@@ -9,14 +9,17 @@ use crate::sync::manager::{BlockProcessType, SyncMessage};
 use crate::sync::{BatchProcessResult, ChainId};
 use beacon_chain::blob_verification::AsBlock;
 use beacon_chain::blob_verification::BlockWrapper;
-use beacon_chain::{AvailabilityProcessingStatus, CountUnrealized};
 use beacon_chain::{
+    observed_block_producers::Error as ObserveError, validator_monitor::get_block_delay_ms,
     BeaconChainError, BeaconChainTypes, BlockError, ChainSegmentResult, HistoricalBlockError,
     NotifyExecutionLayer,
 };
+use beacon_chain::{AvailabilityProcessingStatus, CountUnrealized};
 use lighthouse_network::PeerAction;
 use slog::{debug, error, info, warn};
+use slot_clock::SlotClock;
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::mpsc;
 use types::{Epoch, Hash256, SignedBeaconBlock};
 
@@ -85,6 +88,66 @@ impl<T: BeaconChainTypes> Worker<T> {
                 return;
             }
         };
+
+        // Returns `true` if the time now is after the 4s attestation deadline.
+        let block_is_late = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            // If we can't read the system time clock then indicate that the
+            // block is late (and therefore should *not* be requeued). This
+            // avoids infinite loops.
+            .map_or(true, |now| {
+                get_block_delay_ms(now, block.message(), &self.chain.slot_clock)
+                    > self.chain.slot_clock.unagg_attestation_production_delay()
+            });
+
+        // Checks if a block from this proposer is already known.
+        let proposal_already_known = || {
+            match self
+                .chain
+                .observed_block_producers
+                .read()
+                .proposer_has_been_observed(block.message())
+            {
+                Ok(is_observed) => is_observed,
+                // Both of these blocks will be rejected, so reject them now rather
+                // than re-queuing them.
+                Err(ObserveError::FinalizedBlock { .. })
+                | Err(ObserveError::ValidatorIndexTooHigh { .. }) => false,
+            }
+        };
+
+        // If we've already seen a block from this proposer *and* the block
+        // arrived before the attestation deadline, requeue it to ensure it is
+        // imported late enough that it won't receive a proposer boost.
+        if !block_is_late && 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();