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/network/src/beacon_processor/mod.rs

@@ -41,7 +41,7 @@
 use crate::sync::manager::BlockProcessType;
 use crate::{metrics, service::NetworkMessage, sync::SyncMessage};
 use beacon_chain::parking_lot::Mutex;
-use beacon_chain::{BeaconChain, BeaconChainTypes, GossipVerifiedBlock};
+use beacon_chain::{BeaconChain, BeaconChainTypes, GossipVerifiedBlock, NotifyExecutionLayer};
 use derivative::Derivative;
 use futures::stream::{Stream, StreamExt};
 use futures::task::Poll;
@@ -1587,8 +1587,24 @@ impl<T: BeaconChainTypes> BeaconProcessor<T> {
             /*
              * Verification for a chain segment (multiple blocks).
              */
-            Work::ChainSegment { process_id, blocks } => task_spawner
-                .spawn_async(async move { worker.process_chain_segment(process_id, blocks).await }),
+            Work::ChainSegment { process_id, blocks } => {
+                let notify_execution_layer = if self
+                    .network_globals
+                    .sync_state
+                    .read()
+                    .is_syncing_finalized()
+                {
+                    NotifyExecutionLayer::No
+                } else {
+                    NotifyExecutionLayer::Yes
+                };
+
+                task_spawner.spawn_async(async move {
+                    worker
+                        .process_chain_segment(process_id, blocks, notify_execution_layer)
+                        .await
+                })
+            }
             /*
              * Processing of Status Messages.
              */

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

@@ -7,7 +7,7 @@ use beacon_chain::{
     sync_committee_verification::{self, Error as SyncCommitteeError},
     validator_monitor::get_block_delay_ms,
     BeaconChainError, BeaconChainTypes, BlockError, CountUnrealized, ForkChoiceError,
-    GossipVerifiedBlock,
+    GossipVerifiedBlock, NotifyExecutionLayer,
 };
 use lighthouse_network::{Client, MessageAcceptance, MessageId, PeerAction, PeerId, ReportSource};
 use slog::{crit, debug, error, info, trace, warn};
@@ -934,7 +934,12 @@ impl<T: BeaconChainTypes> Worker<T> {
 
         match self
             .chain
-            .process_block(block_root, verified_block, CountUnrealized::True)
+            .process_block(
+                block_root,
+                verified_block,
+                CountUnrealized::True,
+                NotifyExecutionLayer::Yes,
+            )
             .await
         {
             Ok(block_root) => {

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

@@ -10,6 +10,7 @@ use crate::sync::{BatchProcessResult, ChainId};
 use beacon_chain::CountUnrealized;
 use beacon_chain::{
     BeaconChainError, BeaconChainTypes, BlockError, ChainSegmentResult, HistoricalBlockError,
+    NotifyExecutionLayer,
 };
 use lighthouse_network::PeerAction;
 use slog::{debug, error, info, warn};
@@ -85,7 +86,12 @@ impl<T: BeaconChainTypes> Worker<T> {
         let slot = block.slot();
         let result = self
             .chain
-            .process_block(block_root, block, CountUnrealized::True)
+            .process_block(
+                block_root,
+                block,
+                CountUnrealized::True,
+                NotifyExecutionLayer::Yes,
+            )
             .await;
 
         metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
@@ -127,6 +133,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         &self,
         sync_type: ChainSegmentProcessId,
         downloaded_blocks: Vec<Arc<SignedBeaconBlock<T::EthSpec>>>,
+        notify_execution_layer: NotifyExecutionLayer,
     ) {
         let result = match sync_type {
             // this a request from the range sync
@@ -136,7 +143,11 @@ impl<T: BeaconChainTypes> Worker<T> {
                 let sent_blocks = downloaded_blocks.len();
 
                 match self
-                    .process_blocks(downloaded_blocks.iter(), count_unrealized)
+                    .process_blocks(
+                        downloaded_blocks.iter(),
+                        count_unrealized,
+                        notify_execution_layer,
+                    )
                     .await
                 {
                     (_, Ok(_)) => {
@@ -215,7 +226,11 @@ impl<T: BeaconChainTypes> Worker<T> {
                 // parent blocks are ordered from highest slot to lowest, so we need to process in
                 // reverse
                 match self
-                    .process_blocks(downloaded_blocks.iter().rev(), CountUnrealized::True)
+                    .process_blocks(
+                        downloaded_blocks.iter().rev(),
+                        CountUnrealized::True,
+                        notify_execution_layer,
+                    )
                     .await
                 {
                     (imported_blocks, Err(e)) => {
@@ -246,11 +261,12 @@ impl<T: BeaconChainTypes> Worker<T> {
         &self,
         downloaded_blocks: impl Iterator<Item = &'a Arc<SignedBeaconBlock<T::EthSpec>>>,
         count_unrealized: CountUnrealized,
+        notify_execution_layer: NotifyExecutionLayer,
     ) -> (usize, Result<(), ChainSegmentFailed>) {
         let blocks: Vec<Arc<_>> = downloaded_blocks.cloned().collect();
         match self
             .chain
-            .process_chain_segment(blocks, count_unrealized)
+            .process_chain_segment(blocks, count_unrealized, notify_execution_layer)
             .await
         {
             ChainSegmentResult::Successful { imported_blocks } => {