From 3f4d7f801b1c8b892d1dbc2841b624fadf19f051 Mon Sep 17 00:00:00 2001 From: alok-108 Date: Sun, 24 May 2026 16:12:59 +0530 Subject: [PATCH] feat: implement optimistic sync (#395) - Add skip_execution_validation parameter to on_block - Introduce OptimisticRootsTable for tracking optimistic blocks - Implement process_block_optimistic and handle_invalid_payload - Add is_optimistic_candidate_block helper (SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY = 128) - Extend BlockRangeSyncer with optimistic_mode and new_optimistic constructor - Wire --optimistic-sync CLI flag (default true) into ManagerConfig and service - Hook PayloadStatus::Invalid to handle_invalid_payload for chain backtracking --- bin/ream/src/cli/beacon_node.rs | 4 + .../common/chain/beacon/src/beacon_chain.rs | 74 +++++++++++++++++++ .../beacon/src/electra/beacon_state.rs | 36 ++++++--- .../execution/engine/src/engine_trait.rs | 4 +- crates/common/execution/engine/src/lib.rs | 26 +++++-- .../execution/engine/src/mock_engine.rs | 18 ++++- .../common/fork_choice/beacon/src/handlers.rs | 13 +++- crates/networking/manager/src/config.rs | 1 + crates/networking/manager/src/service.rs | 21 ++++-- .../networking/syncer/src/block_range/mod.rs | 42 ++++++++++- .../syncer/src/block_range/optimistic.rs | 24 ++++++ .../syncer/src/block_range/peer_manager.rs | 17 +++++ crates/storage/src/db/beacon.rs | 7 ++ crates/storage/src/tables/beacon/mod.rs | 1 + .../src/tables/beacon/optimistic_roots.rs | 31 ++++++++ 15 files changed, 284 insertions(+), 35 deletions(-) create mode 100644 crates/networking/syncer/src/block_range/optimistic.rs create mode 100644 crates/storage/src/tables/beacon/optimistic_roots.rs diff --git a/bin/ream/src/cli/beacon_node.rs b/bin/ream/src/cli/beacon_node.rs index 8e57ae3f4..ae1a4bf10 100644 --- a/bin/ream/src/cli/beacon_node.rs +++ b/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, + + #[arg(long, default_value_t = true, help = "Enable optimistic sync")] + pub optimistic_sync: bool, } impl From for ManagerConfig { @@ -108,6 +111,7 @@ impl From 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, } } } diff --git a/crates/common/chain/beacon/src/beacon_chain.rs b/crates/common/chain/beacon/src/beacon_chain.rs index 2e584cf17..64801f8f9 100644 --- a/crates/common/chain/beacon/src/beacon_chain.rs +++ b/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::() { + 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, diff --git a/crates/common/consensus/beacon/src/electra/beacon_state.rs b/crates/common/consensus/beacon/src/electra/beacon_state.rs index 25403448f..3517aa669 100644 --- a/crates/common/consensus/beacon/src/electra/beacon_state.rs +++ b/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 diff --git a/crates/common/execution/engine/src/engine_trait.rs b/crates/common/execution/engine/src/engine_trait.rs index 7bbbb70b8..898ab93f0 100644 --- a/crates/common/execution/engine/src/engine_trait.rs +++ b/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; + ) -> anyhow::Result; async fn engine_get_blobs_v1( &self, diff --git a/crates/common/execution/engine/src/lib.rs b/crates/common/execution/engine/src/lib.rs index e4397cc14..056211d27 100644 --- a/crates/common/execution/engine/src/lib.rs +++ b/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 { + ) -> anyhow::Result { 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 { @@ -426,7 +426,7 @@ impl ExecutionApi for ExecutionEngine { async fn verify_and_notify_new_payload( &self, new_payload_request: NewPayloadRequest, - ) -> anyhow::Result { + ) -> anyhow::Result { let execution_requests_list = get_execution_requests_list(&new_payload_request.execution_requests); if new_payload_request @@ -434,7 +434,11 @@ impl ExecutionApi for ExecutionEngine { .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( @@ -442,14 +446,22 @@ impl ExecutionApi for ExecutionEngine { 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( diff --git a/crates/common/execution/engine/src/mock_engine.rs b/crates/common/execution/engine/src/mock_engine.rs index 9775bde07..c4154f1c0 100644 --- a/crates/common/execution/engine/src/mock_engine.rs +++ b/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 { - Ok(self.execution_valid) + ) -> anyhow::Result { + 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( diff --git a/crates/common/fork_choice/beacon/src/handlers.rs b/crates/common/fork_choice/beacon/src/handlers.rs index e87a26507..5869b0f2c 100644 --- a/crates/common/fork_choice/beacon/src/handlers.rs +++ b/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( store: &mut Store, signed_block: &SignedBeaconBlock, - execution_engine: &Option, + execution_engine: &Option, 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 diff --git a/crates/networking/manager/src/config.rs b/crates/networking/manager/src/config.rs index 289c096f6..4992b1970 100644 --- a/crates/networking/manager/src/config.rs +++ b/crates/networking/manager/src/config.rs @@ -18,4 +18,5 @@ pub struct ManagerConfig { pub enable_builder: bool, pub mev_relay_url: Option, pub blob_retention_epochs: Option, + pub optimistic_sync: bool, } diff --git a/crates/networking/manager/src/service.rs b/crates/networking/manager/src/service.rs index b178a399f..5407736ae 100644 --- a/crates/networking/manager/src/service.rs +++ b/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, diff --git a/crates/networking/syncer/src/block_range/mod.rs b/crates/networking/syncer/src/block_range/mod.rs index 7e8ee1b29..0e8a50fd1 100644 --- a/crates/networking/syncer/src/block_range/mod.rs +++ b/crates/networking/syncer/src/block_range/mod.rs @@ -1,4 +1,5 @@ mod block_cache; +pub mod optimistic; mod peer_manager; mod peer_range_downloader; @@ -39,6 +40,7 @@ pub struct BlockRangeSyncer { pub peer_manager: PeerManager, pub p2p_sender: UnboundedSender, pub executor: ReamExecutor, + pub optimistic_mode: bool, } impl BlockRangeSyncer { @@ -53,6 +55,22 @@ impl BlockRangeSyncer { p2p_sender, peer_manager: PeerManager::new(network_state), executor, + optimistic_mode: false, + } + } + + pub fn new_optimistic( + beacon_chain: Arc, + p2p_sender: UnboundedSender, + network_state: Arc, + executor: ReamExecutor, + ) -> Self { + Self { + beacon_chain, + p2p_sender, + peer_manager: PeerManager::new(network_state), + executor, + optimistic_mode: true, } } @@ -108,17 +126,17 @@ impl BlockRangeSyncer { loop { poll_ready_tasks(&mut task_handles, &mut block_cache, &mut self.peer_manager)?; - let finalized_slot = match self.peer_manager.finalized_slot() { - Some(finalized_slot) => finalized_slot, + let target_slot = match if self.optimistic_mode { self.peer_manager.head_slot() } else { self.peer_manager.finalized_slot() } { + Some(slot) => slot, None => { - warn!("No peers available to determine finalized slot, retrying..."); + warn!("No peers available to determine target slot, retrying..."); sleep(SLEEP_DURATION).await; self.peer_manager.update_peer_set(); continue; } }; - let data_to_fetch = block_cache.data_to_fetch(finalized_slot); + let data_to_fetch = block_cache.data_to_fetch(target_slot); info!( "Forward sync status: Downloaded Blocks {}, Downloaded Blobs {}/{}, Stage {data_to_fetch}", block_cache.block_count(), @@ -223,6 +241,22 @@ impl BlockRangeSyncer { } } + if self.optimistic_mode { + let store = self.beacon_chain.store.lock().await; + let current_head_slot = store.get_current_slot().unwrap_or(0); + let is_optimistic = crate::block_range::optimistic::is_optimistic_candidate_block( + &*store, + current_head_slot, + &block, + ); + drop(store); + + if is_optimistic { + self.beacon_chain.process_block_optimistic(block).await?; + continue; + } + } + self.beacon_chain.process_block(block).await?; } diff --git a/crates/networking/syncer/src/block_range/optimistic.rs b/crates/networking/syncer/src/block_range/optimistic.rs new file mode 100644 index 000000000..d4579f807 --- /dev/null +++ b/crates/networking/syncer/src/block_range/optimistic.rs @@ -0,0 +1,24 @@ +use ream_consensus_beacon::electra::beacon_block::SignedBeaconBlock; +use ream_fork_choice_beacon::store::Store; + +pub const SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY: u64 = 128; + +/// Check if a block can be optimistically imported. +/// Returns true if the block's parent has an execution payload, +/// or if the block is within 128 slots of the current head. +pub fn is_optimistic_candidate_block( + store: &Store, + current_head_slot: u64, + block: &SignedBeaconBlock, +) -> bool { + // If parent has execution payload, we can optimistic import + let parent_root = block.message.parent_root; + if let Ok(Some(_parent_block)) = store.db.block_provider().get(parent_root) { + // Electra blocks always have an execution payload + return true; + } + + // Within safe distance from head? + let distance = current_head_slot.saturating_sub(block.message.slot); + distance <= SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY +} diff --git a/crates/networking/syncer/src/block_range/peer_manager.rs b/crates/networking/syncer/src/block_range/peer_manager.rs index 67ca9fbd9..48f3a37fa 100644 --- a/crates/networking/syncer/src/block_range/peer_manager.rs +++ b/crates/networking/syncer/src/block_range/peer_manager.rs @@ -114,4 +114,21 @@ impl PeerManager { .max_by_key(|&(_, count)| count) .map(|(slot, _)| slot) } + + pub fn head_slot(&self) -> Option { + let mut frequencies = HashMap::new(); + + for peer in self.peers.values() { + if let Some(status) = &peer.peer.status { + *frequencies + .entry(status.head_slot) + .or_insert(0) += 1; + } + } + + frequencies + .into_iter() + .max_by_key(|&(_, count)| count) + .map(|(slot, _)| slot) + } } diff --git a/crates/storage/src/db/beacon.rs b/crates/storage/src/db/beacon.rs index c7b9761b6..67e79717a 100644 --- a/crates/storage/src/db/beacon.rs +++ b/crates/storage/src/db/beacon.rs @@ -21,6 +21,7 @@ use crate::{ unrealized_finalized_checkpoint::UnrealizedFinalizedCheckpointField, unrealized_justifications::UnrealizedJustificationsTable, unrealized_justified_checkpoint::UnrealizedJustifiedCheckpointField, + optimistic_roots::OptimisticRootsTable, }, table::REDBTable, }, @@ -54,6 +55,12 @@ impl BeaconDB { } } + pub fn optimistic_roots_provider(&self) -> OptimisticRootsTable { + OptimisticRootsTable { + db: self.db.clone(), + } + } + pub fn blobs_and_proofs_provider(&self) -> BlobsAndProofsTable { BlobsAndProofsTable { data_dir: self.data_dir.clone(), diff --git a/crates/storage/src/tables/beacon/mod.rs b/crates/storage/src/tables/beacon/mod.rs index 1c06641d1..7bbf8f73c 100644 --- a/crates/storage/src/tables/beacon/mod.rs +++ b/crates/storage/src/tables/beacon/mod.rs @@ -1,4 +1,5 @@ pub mod beacon_block; +pub mod optimistic_roots; pub mod beacon_state; pub mod blobs_and_proofs; pub mod block_timeliness; diff --git a/crates/storage/src/tables/beacon/optimistic_roots.rs b/crates/storage/src/tables/beacon/optimistic_roots.rs new file mode 100644 index 000000000..6c10cf5cd --- /dev/null +++ b/crates/storage/src/tables/beacon/optimistic_roots.rs @@ -0,0 +1,31 @@ +use std::sync::Arc; + +use alloy_primitives::B256; +use redb::{Database, TableDefinition}; + +use crate::tables::{ssz_encoder::SSZEncoding, table::REDBTable}; + +pub struct OptimisticRootsTable { + pub db: Arc, +} + +/// Table definition for the Optimistic Roots table +/// +/// Key: block_root +/// Value: bool +impl REDBTable for OptimisticRootsTable { + const TABLE_DEFINITION: TableDefinition<'_, SSZEncoding, SSZEncoding> = + TableDefinition::new("optimistic_roots"); + + type Key = B256; + + type KeyTableDefinition = SSZEncoding; + + type Value = bool; + + type ValueTableDefinition = SSZEncoding; + + fn database(&self) -> Arc { + self.db.clone() + } +}