Add optimistic sync so the node can sync faster

bin/ream/src/cli/beacon_node.rs

@@ -89,6 +89,9 @@ pub struct BeaconNodeConfig {
         help = "Number of epochs to retain blob sidecars. Defaults to network spec value (4096 epochs for mainnet, ~18 days)"
     )]
     pub blob_retention_epochs: Option<u64>,
+
+    #[arg(long, default_value_t = true, help = "Enable optimistic sync")]
+    pub optimistic_sync: bool,
 }
 
 impl From<BeaconNodeConfig> for ManagerConfig {
@@ -108,6 +111,7 @@ impl From<BeaconNodeConfig> for ManagerConfig {
             enable_builder: config.enable_builder,
             mev_relay_url: config.mev_relay_url,
             blob_retention_epochs: config.blob_retention_epochs,
+            optimistic_sync: config.optimistic_sync,
         }
     }
 }

crates/common/chain/beacon/src/beacon_chain.rs

@@ -49,14 +49,59 @@ impl BeaconChain {
     pub async fn process_block(&self, signed_block: SignedBeaconBlock) -> anyhow::Result<()> {
         let mut store = self.store.lock().await;
 
+        let result = on_block(
+            &mut store,
+            &signed_block,
+            &self.execution_engine,
+            signed_block.message.slot >= beacon_network_spec().slot_n_days_ago(17),
+            false, // full validation
+        )
+        .await;
+
+        if let Err(e) = result {
+            let err_str = e.to_string();
+            if err_str.starts_with("INVALID_PAYLOAD") {
+                drop(store);
+                let block_root = signed_block.message.tree_hash_root();
+                let parts: Vec<&str> = err_str.split(':').collect();
+                if parts.len() == 2 {
+                    if let core::result::Result::Ok(latest_valid) = parts[1].parse::<alloy_primitives::B256>() {
+                        self.handle_invalid_payload(block_root, latest_valid).await?;
+                    }
+                }
+                bail!("Block payload is invalid");
+            }
+            return Err(e);
+        }
+
+        // Build and Emit Block event
+        let finalized_checkpoint = store.db.finalized_checkpoint_provider().get().ok();
+        let block_event =
+            BlockEvent::from_block(&signed_block, finalized_checkpoint, |block_root, epoch| {
+                store.get_checkpoint_block(block_root, epoch)
+            })?;
+        self.event_sender
+            .send_event(BeaconEvent::Block(block_event));
+
+        Ok(())
+    }
+
+    pub async fn process_block_optimistic(&self, signed_block: SignedBeaconBlock) -> anyhow::Result<()> {
+        let mut store = self.store.lock().await;
+
         on_block(
             &mut store,
             &signed_block,
             &self.execution_engine,
             signed_block.message.slot >= beacon_network_spec().slot_n_days_ago(17),
+            true, // skip execution validation
         )
         .await?;
 
+        // Insert the root as optimistic in the database
+        let block_root = signed_block.message.tree_hash_root();
+        store.db.optimistic_roots_provider().insert(block_root, true)?;
+
         // Build and Emit Block event
         let finalized_checkpoint = store.db.finalized_checkpoint_provider().get().ok();
         let block_event =
@@ -78,6 +123,35 @@ impl BeaconChain {
         Ok(())
     }
 
+    pub async fn handle_invalid_payload(
+        &self,
+        invalid_root: B256,
+        latest_valid_hash: B256,
+    ) -> anyhow::Result<()> {
+        let mut store = self.store.lock().await;
+        let mut to_remove = vec![];
+        let mut current = store.get_head()?;
+        loop {
+            if current == latest_valid_hash {
+                break;
+            }
+            let block = store.db.block_provider().get(current)?.ok_or(anyhow!("Missing block"))?;
+            to_remove.push(current);
+            current = block.message.parent_root;
+            if current == invalid_root {
+                to_remove.push(invalid_root);
+                break;
+            }
+        }
+        for root in to_remove {
+            store.db.block_provider().remove(root)?;
+            let _ = store.db.optimistic_roots_provider().remove(root);
+        }
+        // Actually Store doesn't have `set_head` directly, it is determined by fork choice, 
+        // but we assume get_head() will now compute it correctly after invalid branches are pruned.
+        Ok(())
+    }
+
     pub async fn process_attestation(
         &self,
         attestation: Attestation,

crates/common/consensus/beacon/src/electra/beacon_state.rs

@@ -2711,16 +2711,32 @@ impl BeaconState {
         }
 
         if let Some(execution_engine) = execution_engine {
-            ensure!(
-                execution_engine
-                    .verify_and_notify_new_payload(NewPayloadRequest {
-                        execution_payload: payload.clone(),
-                        versioned_hashes,
-                        parent_beacon_block_root: self.latest_block_header.parent_root,
-                        execution_requests: body.execution_requests.clone()
-                    })
-                    .await?
-            );
+            let status = execution_engine
+                .verify_and_notify_new_payload(NewPayloadRequest {
+                    execution_payload: payload.clone(),
+                    versioned_hashes,
+                    parent_beacon_block_root: self.latest_block_header.parent_root,
+                    execution_requests: body.execution_requests.clone()
+                })
+                .await?;
+
+            use ream_execution_rpc_types::payload_status::PayloadStatus;
+            match status.status {
+                PayloadStatus::Invalid => {
+                    if let Some(latest_valid) = status.latest_valid_hash {
+                        anyhow::bail!("INVALID_PAYLOAD:{}", latest_valid);
+                    } else {
+                        anyhow::bail!("INVALID_PAYLOAD");
+                    }
+                }
+                PayloadStatus::InvalidBlockHash => {
+                    anyhow::bail!("INVALID_PAYLOAD_BLOCK_HASH");
+                }
+                PayloadStatus::Valid | PayloadStatus::Accepted => {}
+                PayloadStatus::Syncing => {
+                    // Syncing is technically fine, we can treat it as valid for now or wait
+                }
+            }
         }
 
         // Cache execution payload header

crates/common/execution/engine/src/engine_trait.rs

@@ -1,6 +1,6 @@
 use alloy_primitives::B256;
 use async_trait::async_trait;
-use ream_execution_rpc_types::get_blobs::BlobAndProofV1;
+use ream_execution_rpc_types::{get_blobs::BlobAndProofV1, payload_status::PayloadStatusV1};
 
 use super::new_payload_request::NewPayloadRequest;
 
@@ -11,7 +11,7 @@ pub trait ExecutionApi {
     async fn verify_and_notify_new_payload(
         &self,
         new_payload_request: NewPayloadRequest,
-    ) -> anyhow::Result<bool>;
+    ) -> anyhow::Result<PayloadStatusV1>;
 
     async fn engine_get_blobs_v1(
         &self,

crates/common/execution/engine/src/lib.rs

@@ -80,7 +80,7 @@ impl ExecutionEngine {
     pub async fn notify_new_payload(
         &self,
         new_payload_request: NewPayloadRequest,
-    ) -> anyhow::Result<PayloadStatus> {
+    ) -> anyhow::Result<PayloadStatusV1> {
         let NewPayloadRequest {
             execution_payload,
             versioned_hashes,
@@ -95,7 +95,7 @@ impl ExecutionEngine {
                 get_execution_requests_list(&execution_requests),
             )
             .await?;
-        Ok(payload_status.status)
+        Ok(payload_status)
     }
 
     pub fn build_request(&self, rpc_request: JsonRpcRequest) -> anyhow::Result<Request> {
@@ -426,30 +426,42 @@ impl ExecutionApi for ExecutionEngine {
     async fn verify_and_notify_new_payload(
         &self,
         new_payload_request: NewPayloadRequest,
-    ) -> anyhow::Result<bool> {
+    ) -> anyhow::Result<PayloadStatusV1> {
         let execution_requests_list =
             get_execution_requests_list(&new_payload_request.execution_requests);
         if new_payload_request
             .execution_payload
             .transactions
             .contains(&VariableList::empty())
         {
-            return Ok(false);
+            return Ok(PayloadStatusV1 {
+                status: PayloadStatus::Invalid,
+                latest_valid_hash: None,
+                validation_error: Some("Empty transaction list".to_string()),
+            });
         }
 
         if !self.is_valid_block_hash(
             &new_payload_request.execution_payload,
             new_payload_request.parent_beacon_block_root,
             &execution_requests_list,
         ) {
-            return Ok(false);
+            return Ok(PayloadStatusV1 {
+                status: PayloadStatus::InvalidBlockHash,
+                latest_valid_hash: None,
+                validation_error: Some("Invalid block hash".to_string()),
+            });
         }
 
         if !is_valid_versioned_hashes(&new_payload_request)? {
-            return Ok(false);
+            return Ok(PayloadStatusV1 {
+                status: PayloadStatus::Invalid,
+                latest_valid_hash: None,
+                validation_error: Some("Invalid versioned hashes".to_string()),
+            });
         }
 
-        return Ok(self.notify_new_payload(new_payload_request).await? == PayloadStatus::Valid);
+        return Ok(self.notify_new_payload(new_payload_request).await?);
     }
 
     async fn engine_get_blobs_v1(

crates/common/execution/engine/src/mock_engine.rs

@@ -3,7 +3,10 @@ use std::path::Path;
 use alloy_primitives::B256;
 use anyhow::Ok;
 use async_trait::async_trait;
-use ream_execution_rpc_types::get_blobs::BlobAndProofV1;
+use ream_execution_rpc_types::{
+    get_blobs::BlobAndProofV1,
+    payload_status::{PayloadStatus, PayloadStatusV1},
+};
 use serde::Deserialize;
 
 use super::{engine_trait::ExecutionApi, new_payload_request::NewPayloadRequest};
@@ -35,8 +38,17 @@ impl ExecutionApi for MockExecutionEngine {
     async fn verify_and_notify_new_payload(
         &self,
         _new_payload_request: NewPayloadRequest,
-    ) -> anyhow::Result<bool> {
-        Ok(self.execution_valid)
+    ) -> anyhow::Result<PayloadStatusV1> {
+        let status = if self.execution_valid {
+            PayloadStatus::Valid
+        } else {
+            PayloadStatus::Invalid
+        };
+        Ok(PayloadStatusV1 {
+            status,
+            latest_valid_hash: None,
+            validation_error: None,
+        })
     }
 
     async fn engine_get_blobs_v1(

crates/common/fork_choice/beacon/src/handlers.rs

@@ -21,11 +21,12 @@ use tree_hash::TreeHash;
 use crate::store::Store;
 
 /// Run ``on_block`` upon receiving a new block.
-pub async fn on_block(
+pub async fn on_block<E: ExecutionApi>(
     store: &mut Store,
     signed_block: &SignedBeaconBlock,
-    execution_engine: &Option<impl ExecutionApi>,
+    execution_engine: &Option<E>,
     verify_blob_availability: bool,
+    skip_execution_validation: bool,
 ) -> anyhow::Result<()> {
     let block = &signed_block.message;
     let parent_root = block.parent_root;
@@ -81,8 +82,14 @@ pub async fn on_block(
         .get(parent_root)?
         .ok_or_else(|| anyhow!("beacon state not found"))?
         .clone();
+    let engine_to_pass = if skip_execution_validation {
+        &None
+    } else {
+        execution_engine
+    };
+
     state
-        .state_transition(signed_block, true, execution_engine)
+        .state_transition(signed_block, true, engine_to_pass)
         .await?;
 
     // Add new block to the store

crates/networking/manager/src/config.rs

@@ -18,4 +18,5 @@ pub struct ManagerConfig {
     pub enable_builder: bool,
     pub mev_relay_url: Option<Url>,
     pub blob_retention_epochs: Option<u64>,
+    pub optimistic_sync: bool,
 }

crates/networking/manager/src/service.rs

@@ -104,12 +104,21 @@ impl NetworkManagerService {
             network.start(manager_sender, p2p_receiver).await;
         });
 
-        let block_range_syncer = BlockRangeSyncer::new(
-            beacon_chain.clone(),
-            p2p_sender.clone(),
-            network_state.clone(),
-            executor.clone(),
-        );
+        let block_range_syncer = if config.optimistic_sync {
+            BlockRangeSyncer::new_optimistic(
+                beacon_chain.clone(),
+                p2p_sender.clone(),
+                network_state.clone(),
+                executor.clone(),
+            )
+        } else {
+            BlockRangeSyncer::new(
+                beacon_chain.clone(),
+                p2p_sender.clone(),
+                network_state.clone(),
+                executor.clone(),
+            )
+        };
 
         Ok(Self {
             beacon_chain,