Optimize finalized chain sync by skipping newPayload messages (#3738)

## Issue Addressed

#3704 

## Proposed Changes
Adds is_syncing_finalized: bool parameter for block verification functions. Sets the payload_verification_status to Optimistic if is_syncing_finalized is true. Uses SyncState in NetworkGlobals in BeaconProcessor to retrieve the syncing status.

## Additional Info
I could implement FinalizedSignatureVerifiedBlock if you think it would be nicer.

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -18,7 +18,7 @@ use crate::errors::{BeaconChainError as Error, BlockProductionError};
 use crate::eth1_chain::{Eth1Chain, Eth1ChainBackend};
 use crate::eth1_finalization_cache::{Eth1FinalizationCache, Eth1FinalizationData};
 use crate::events::ServerSentEventHandler;
-use crate::execution_payload::{get_execution_payload, PreparePayloadHandle};
+use crate::execution_payload::{get_execution_payload, NotifyExecutionLayer, PreparePayloadHandle};
 use crate::fork_choice_signal::{ForkChoiceSignalRx, ForkChoiceSignalTx, ForkChoiceWaitResult};
 use crate::head_tracker::HeadTracker;
 use crate::historical_blocks::HistoricalBlockError;
@@ -2341,6 +2341,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self: &Arc<Self>,
         chain_segment: Vec<Arc<SignedBeaconBlock<T::EthSpec>>>,
         count_unrealized: CountUnrealized,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> ChainSegmentResult<T::EthSpec> {
         let mut imported_blocks = 0;
 
@@ -2409,6 +2410,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         signature_verified_block.block_root(),
                         signature_verified_block,
                         count_unrealized,
+                        notify_execution_layer,
                     )
                     .await
                 {
@@ -2497,6 +2499,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block_root: Hash256,
         unverified_block: B,
         count_unrealized: CountUnrealized,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<Hash256, BlockError<T::EthSpec>> {
         // Start the Prometheus timer.
         let _full_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_TIMES);
@@ -2510,8 +2513,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // A small closure to group the verification and import errors.
         let chain = self.clone();
         let import_block = async move {
-            let execution_pending =
-                unverified_block.into_execution_pending_block(block_root, &chain)?;
+            let execution_pending = unverified_block.into_execution_pending_block(
+                block_root,
+                &chain,
+                notify_execution_layer,
+            )?;
             chain
                 .import_execution_pending_block(execution_pending, count_unrealized)
                 .await

beacon_node/beacon_chain/src/block_verification.rs

@@ -45,7 +45,7 @@
 use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::execution_payload::{
     is_optimistic_candidate_block, validate_execution_payload_for_gossip, validate_merge_block,
-    AllowOptimisticImport, PayloadNotifier,
+    AllowOptimisticImport, NotifyExecutionLayer, PayloadNotifier,
 };
 use crate::snapshot_cache::PreProcessingSnapshot;
 use crate::validator_monitor::HISTORIC_EPOCHS as VALIDATOR_MONITOR_HISTORIC_EPOCHS;
@@ -636,8 +636,9 @@ pub trait IntoExecutionPendingBlock<T: BeaconChainTypes>: Sized {
         self,
         block_root: Hash256,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<ExecutionPendingBlock<T>, BlockError<T::EthSpec>> {
-        self.into_execution_pending_block_slashable(block_root, chain)
+        self.into_execution_pending_block_slashable(block_root, chain, notify_execution_layer)
             .map(|execution_pending| {
                 // Supply valid block to slasher.
                 if let Some(slasher) = chain.slasher.as_ref() {
@@ -653,6 +654,7 @@ pub trait IntoExecutionPendingBlock<T: BeaconChainTypes>: Sized {
         self,
         block_root: Hash256,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<ExecutionPendingBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>>;
 
     fn block(&self) -> &SignedBeaconBlock<T::EthSpec>;
@@ -899,10 +901,15 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for GossipVerifiedBlock<T
         self,
         block_root: Hash256,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<ExecutionPendingBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
         let execution_pending =
             SignatureVerifiedBlock::from_gossip_verified_block_check_slashable(self, chain)?;
-        execution_pending.into_execution_pending_block_slashable(block_root, chain)
+        execution_pending.into_execution_pending_block_slashable(
+            block_root,
+            chain,
+            notify_execution_layer,
+        )
     }
 
     fn block(&self) -> &SignedBeaconBlock<T::EthSpec> {
@@ -1032,6 +1039,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for SignatureVerifiedBloc
         self,
         block_root: Hash256,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<ExecutionPendingBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
         let header = self.block.signed_block_header();
         let (parent, block) = if let Some(parent) = self.parent {
@@ -1047,6 +1055,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for SignatureVerifiedBloc
             parent,
             self.consensus_context,
             chain,
+            notify_execution_layer,
         )
         .map_err(|e| BlockSlashInfo::SignatureValid(header, e))
     }
@@ -1063,13 +1072,14 @@ impl<T: BeaconChainTypes> IntoExecutionPendingBlock<T> for Arc<SignedBeaconBlock
         self,
         block_root: Hash256,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<ExecutionPendingBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
         // Perform an early check to prevent wasting time on irrelevant blocks.
         let block_root = check_block_relevancy(&self, block_root, chain)
             .map_err(|e| BlockSlashInfo::SignatureNotChecked(self.signed_block_header(), e))?;
 
         SignatureVerifiedBlock::check_slashable(self, block_root, chain)?
-            .into_execution_pending_block_slashable(block_root, chain)
+            .into_execution_pending_block_slashable(block_root, chain, notify_execution_layer)
     }
 
     fn block(&self) -> &SignedBeaconBlock<T::EthSpec> {
@@ -1091,6 +1101,7 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
         parent: PreProcessingSnapshot<T::EthSpec>,
         mut consensus_context: ConsensusContext<T::EthSpec>,
         chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<Self, BlockError<T::EthSpec>> {
         if let Some(parent) = chain
             .canonical_head
@@ -1237,7 +1248,8 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
 
         // Define a future that will verify the execution payload with an execution engine (but
         // don't execute it yet).
-        let payload_notifier = PayloadNotifier::new(chain.clone(), block.clone(), &state)?;
+        let payload_notifier =
+            PayloadNotifier::new(chain.clone(), block.clone(), &state, notify_execution_layer)?;
         let is_valid_merge_transition_block =
             is_merge_transition_block(&state, block.message().body());
         let payload_verification_future = async move {

beacon_node/beacon_chain/src/execution_payload.rs

@@ -35,6 +35,16 @@ pub enum AllowOptimisticImport {
     No,
 }
 
+/// Signal whether the execution payloads of new blocks should be
+/// immediately verified with the EL or imported optimistically without
+/// any EL communication.
+#[derive(Default, Clone, Copy)]
+pub enum NotifyExecutionLayer {
+    #[default]
+    Yes,
+    No,
+}
+
 /// Used to await the result of executing payload with a remote EE.
 pub struct PayloadNotifier<T: BeaconChainTypes> {
     pub chain: Arc<BeaconChain<T>>,
@@ -47,21 +57,27 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
         chain: Arc<BeaconChain<T>>,
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
         state: &BeaconState<T::EthSpec>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> Result<Self, BlockError<T::EthSpec>> {
-        let payload_verification_status = if is_execution_enabled(state, block.message().body()) {
-            // Perform the initial stages of payload verification.
-            //
-            // We will duplicate these checks again during `per_block_processing`, however these checks
-            // are cheap and doing them here ensures we protect the execution engine from junk.
-            partially_verify_execution_payload(
-                state,
-                block.message().execution_payload()?,
-                &chain.spec,
-            )
-            .map_err(BlockError::PerBlockProcessingError)?;
-            None
-        } else {
-            Some(PayloadVerificationStatus::Irrelevant)
+        let payload_verification_status = match notify_execution_layer {
+            NotifyExecutionLayer::No => Some(PayloadVerificationStatus::Optimistic),
+            NotifyExecutionLayer::Yes => {
+                if is_execution_enabled(state, block.message().body()) {
+                    // Perform the initial stages of payload verification.
+                    //
+                    // We will duplicate these checks again during `per_block_processing`, however these checks
+                    // are cheap and doing them here ensures we protect the execution engine from junk.
+                    partially_verify_execution_payload(
+                        state,
+                        block.message().execution_payload()?,
+                        &chain.spec,
+                    )
+                    .map_err(BlockError::PerBlockProcessingError)?;
+                    None
+                } else {
+                    Some(PayloadVerificationStatus::Irrelevant)
+                }
+            }
         };
 
         Ok(Self {

beacon_node/beacon_chain/src/lib.rs

@@ -63,6 +63,7 @@ pub use canonical_head::{CachedHead, CanonicalHead, CanonicalHeadRwLock};
 pub use eth1_chain::{Eth1Chain, Eth1ChainBackend};
 pub use events::ServerSentEventHandler;
 pub use execution_layer::EngineState;
+pub use execution_payload::NotifyExecutionLayer;
 pub use fork_choice::{ExecutionStatus, ForkchoiceUpdateParameters};
 pub use metrics::scrape_for_metrics;
 pub use parking_lot;

beacon_node/beacon_chain/src/test_utils.rs

@@ -2,7 +2,7 @@ pub use crate::persisted_beacon_chain::PersistedBeaconChain;
 pub use crate::{
     beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY},
     migrate::MigratorConfig,
-    BeaconChainError, ProduceBlockVerification,
+    BeaconChainError, NotifyExecutionLayer, ProduceBlockVerification,
 };
 use crate::{
     builder::{BeaconChainBuilder, Witness},
@@ -1460,7 +1460,12 @@ where
         self.set_current_slot(slot);
         let block_hash: SignedBeaconBlockHash = self
             .chain
-            .process_block(block_root, Arc::new(block), CountUnrealized::True)
+            .process_block(
+                block_root,
+                Arc::new(block),
+                CountUnrealized::True,
+                NotifyExecutionLayer::Yes,
+            )
             .await?
             .into();
         self.chain.recompute_head_at_current_slot().await;
@@ -1477,6 +1482,7 @@ where
                 block.canonical_root(),
                 Arc::new(block),
                 CountUnrealized::True,
+                NotifyExecutionLayer::Yes,
             )
             .await?
             .into();