diff --git a/Cargo.lock b/Cargo.lock index 965139d3c68..93ac70fb8d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4338,6 +4338,8 @@ dependencies = [ "hex-literal 0.4.1", "jemalloc_pprof", "jsonwebtoken", + "libssz", + "libssz-derive", "libssz-merkle", "libssz-types", "rand 0.8.5", diff --git a/crates/blockchain/blockchain.rs b/crates/blockchain/blockchain.rs index ba992e1ebdd..1dd18ff93e2 100644 --- a/crates/blockchain/blockchain.rs +++ b/crates/blockchain/blockchain.rs @@ -61,7 +61,7 @@ use crossbeam::channel::{self as cb, TryRecvError, select}; #[cfg(feature = "c-kzg")] use ethrex_common::types::EIP4844Transaction; use ethrex_common::types::block_access_list::BlockAccessList; -use ethrex_common::types::block_execution_witness::ExecutionWitness; +use ethrex_common::types::block_execution_witness::{ExecutionWitness, RpcExecutionWitness}; use ethrex_common::types::fee_config::FeeConfig; use ethrex_common::types::{ AccountInfo, AccountState, AccountUpdate, BalSynthesisItem, Block, BlockHash, BlockHeader, @@ -97,13 +97,13 @@ use payload::PayloadOrTask; use rustc_hash::{FxHashMap, FxHashSet}; use std::collections::hash_map::Entry; use std::collections::{BTreeMap, HashMap, HashSet}; -use std::sync::LazyLock; use std::sync::mpsc::Sender; use std::sync::{ Arc, RwLock, atomic::{AtomicBool, AtomicUsize, Ordering}, mpsc::{Receiver, channel}, }; +use std::sync::{LazyLock, Mutex}; use std::time::{Duration, Instant}; use tokio::sync::Mutex as TokioMutex; use tokio_util::sync::CancellationToken; @@ -213,6 +213,11 @@ pub struct Blockchain { /// Persistent thread pool for merkleization workers. /// 17 threads: 16 shard workers + 1 watcher/coordination. merkle_pool: rayon::ThreadPool, + /// Lock to ensure the previous block's background store completes before the + /// next block's execution begins. Held by the background store thread during + /// `store_witness` + `store_block`, and acquired at the start of the next + /// `add_block_pipeline_inner` call. + store_lock: Arc>, } /// Configuration options for the blockchain. @@ -344,6 +349,7 @@ impl Blockchain { payloads: Arc::new(TokioMutex::new(Vec::new())), options: blockchain_opts, merkle_pool: Self::build_merkle_pool(), + store_lock: Arc::new(Mutex::new(())), } } @@ -355,6 +361,7 @@ impl Blockchain { payloads: Arc::new(TokioMutex::new(Vec::new())), options: BlockchainOptions::default(), merkle_pool: Self::build_merkle_pool(), + store_lock: Arc::new(Mutex::new(())), } } @@ -1853,7 +1860,18 @@ impl Blockchain { block: Block, bal: Option<&BlockAccessList>, ) -> Result<(), ChainError> { - let (_, result) = self.add_block_pipeline_inner(block, bal)?; + let (_, result) = self.add_block_pipeline_inner(block, bal, false)?; + result.map(|_| ()) + } + + /// Same as [`add_block_pipeline`] but also generates and returns an + /// `RpcExecutionWitness` alongside the `PayloadStatus`. + pub fn add_block_pipeline_with_witness( + &self, + block: Block, + bal: Option<&BlockAccessList>, + ) -> Result, ChainError> { + let (_, result) = self.add_block_pipeline_with_witness_inner(block, bal, true)?; result } @@ -1864,11 +1882,180 @@ impl Blockchain { block: Block, bal: Option<&BlockAccessList>, ) -> Result, ChainError> { - let (produced_bal, result) = self.add_block_pipeline_inner(block, bal)?; + let (produced_bal, result) = self.add_block_pipeline_inner(block, bal, false)?; result?; Ok(produced_bal) } + /// A static version of `store_block` that can be used for background task and doesn't hold a reference to `self`. + fn store_block_static( + storage: &Store, + block: Block, + account_updates_list: AccountUpdatesList, + execution_result: BlockExecutionResult, + ) -> Result<(), ChainError> { + validate_state_root(&block.header, account_updates_list.state_trie_hash)?; + + let update_batch = UpdateBatch { + account_updates: account_updates_list.state_updates, + storage_updates: account_updates_list.storage_updates, + receipts: vec![(block.hash(), execution_result.receipts)], + blocks: vec![block], + code_updates: account_updates_list.code_updates, + batch_mode: false, + }; + + storage + .store_block_updates(update_batch) + .map_err(|e| e.into()) + } + + /// This is similar to `add_block_pipeline_inner` but optimized for witness generation. + fn add_block_pipeline_with_witness_inner( + &self, + block: Block, + bal: Option<&BlockAccessList>, + compute_witness: bool, + ) -> Result< + ( + Option, + Result, ChainError>, + ), + ChainError, + > { + // Wait for the previous block's background store to complete before + // starting execution. This ensures trie state is consistent. + let _prev_store = self + .store_lock + .lock() + .map_err(|_| ChainError::Custom("store_lock poisoned".to_string()))?; + drop(_prev_store); + + // Validate if it can be the new head and find the parent + let Ok(parent_header) = find_parent_header(&block.header, &self.storage) else { + // If the parent is not present, we store it as pending. + self.storage.add_pending_block(block)?; + return Err(ChainError::ParentNotFound); + }; + + let (mut vm, logger) = + if (self.options.precompute_witnesses || compute_witness) && self.is_synced() { + // If witness pre-generation is enabled, we wrap the db with a logger + // to track state access (block hashes, storage keys, codes) during execution + // avoiding the need to re-execute the block later. + let vm_db: DynVmDatabase = Box::new(StoreVmDatabase::new( + self.storage.clone(), + parent_header.clone(), + )?); + + let logger = Arc::new(DatabaseLogger::new(Arc::new(vm_db))); + + let vm = match self.options.r#type.clone() { + BlockchainType::L1 => { + Evm::new_from_db_for_l1(logger.clone(), Arc::new(NativeCrypto)) + } + BlockchainType::L2(l2_config) => Evm::new_from_db_for_l2( + logger.clone(), + *l2_config.fee_config.read().map_err(|_| { + EvmError::Custom("Fee config lock was poisoned".to_string()) + })?, + Arc::new(NativeCrypto), + ), + }; + (vm, Some(logger)) + } else { + let vm_db = StoreVmDatabase::new(self.storage.clone(), parent_header.clone())?; + let vm = self.new_evm(vm_db)?; + (vm, None) + }; + + let ( + res, + account_updates_list, + accumulated_updates, + produced_bal, + merkle_queue_length, + instants, + warmer_duration, + ) = { self.execute_block_pipeline(&block, &parent_header, &mut vm, bal)? }; + + let (gas_used, gas_limit, block_number, transactions_count) = ( + block.header.gas_used, + block.header.gas_limit, + block.header.number, + block.body.transactions.len(), + ); + + let mut produced_witness: Option = None; + + if let Some(logger) = logger + && let Some(account_updates) = accumulated_updates + { + let witness = self.generate_witness_from_account_updates( + account_updates, + &block, + parent_header, + &logger, + )?; + + produced_witness = Some(witness); + }; + + // At this point the witness is ready to be returned, for the next set of actions, we would love to do a spawn and forget + // for - storing the witness + // for - storing the block (this is the most compute intensive part) + let store_lock = self.store_lock.clone(); + let produced_witness_clone = produced_witness.clone(); + let storage = self.storage.clone(); + let perf_logs_enabled = self.options.perf_logs_enabled; + + std::thread::spawn(move || { + let _lock = store_lock.lock().unwrap_or_else(|e| e.into_inner()); + + if let Some(ref witness) = produced_witness_clone { + let block_hash = block.hash(); + if let Err(e) = storage.store_witness(block_hash, block_number, witness.clone()) { + ::tracing::warn!("Background store_witness failed: {e}"); + } + } + + if let Err(e) = + Blockchain::store_block_static(&storage, block, account_updates_list, res) + { + ::tracing::error!("Background store_block failed: {e}"); + } + + let stored = Instant::now(); + + let instants = std::array::from_fn(move |i| { + if i < instants.len() { + instants[i] + } else { + stored + } + }); + + if perf_logs_enabled { + Self::print_add_block_pipeline_logs( + gas_used, + gas_limit, + block_number, + transactions_count, + merkle_queue_length, + warmer_duration, + instants, + ); + } + }); + + let witness = + produced_witness.ok_or(ChainError::Custom("No witness produced".to_string()))?; + let rpc_witness = RpcExecutionWitness::try_from(witness) + .map_err(|e| ChainError::Custom(format!("witness conversion failed: {e}")))?; + + Ok((produced_bal, Ok(Some(rpc_witness)))) + } + /// Runs the full block pipeline (execute + merkleize + store). /// /// Returns a two-level Result: @@ -1881,7 +2068,14 @@ impl Blockchain { &self, block: Block, bal: Option<&BlockAccessList>, - ) -> Result<(Option, Result<(), ChainError>), ChainError> { + compute_witness: bool, + ) -> Result< + ( + Option, + Result, ChainError>, + ), + ChainError, + > { // Validate if it can be the new head and find the parent let Ok(parent_header) = find_parent_header(&block.header, &self.storage) else { // If the parent is not present, we store it as pending. @@ -1889,35 +2083,36 @@ impl Blockchain { return Err(ChainError::ParentNotFound); }; - let (mut vm, logger) = if self.options.precompute_witnesses && self.is_synced() { - // If witness pre-generation is enabled, we wrap the db with a logger - // to track state access (block hashes, storage keys, codes) during execution - // avoiding the need to re-execute the block later. - let vm_db: DynVmDatabase = Box::new(StoreVmDatabase::new( - self.storage.clone(), - parent_header.clone(), - )?); - - let logger = Arc::new(DatabaseLogger::new(Arc::new(vm_db))); - - let vm = match self.options.r#type.clone() { - BlockchainType::L1 => { - Evm::new_from_db_for_l1(logger.clone(), Arc::new(NativeCrypto)) - } - BlockchainType::L2(l2_config) => Evm::new_from_db_for_l2( - logger.clone(), - *l2_config.fee_config.read().map_err(|_| { - EvmError::Custom("Fee config lock was poisoned".to_string()) - })?, - Arc::new(NativeCrypto), - ), + let (mut vm, logger) = + if (self.options.precompute_witnesses || compute_witness) && self.is_synced() { + // If witness pre-generation is enabled, we wrap the db with a logger + // to track state access (block hashes, storage keys, codes) during execution + // avoiding the need to re-execute the block later. + let vm_db: DynVmDatabase = Box::new(StoreVmDatabase::new( + self.storage.clone(), + parent_header.clone(), + )?); + + let logger = Arc::new(DatabaseLogger::new(Arc::new(vm_db))); + + let vm = match self.options.r#type.clone() { + BlockchainType::L1 => { + Evm::new_from_db_for_l1(logger.clone(), Arc::new(NativeCrypto)) + } + BlockchainType::L2(l2_config) => Evm::new_from_db_for_l2( + logger.clone(), + *l2_config.fee_config.read().map_err(|_| { + EvmError::Custom("Fee config lock was poisoned".to_string()) + })?, + Arc::new(NativeCrypto), + ), + }; + (vm, Some(logger)) + } else { + let vm_db = StoreVmDatabase::new(self.storage.clone(), parent_header.clone())?; + let vm = self.new_evm(vm_db)?; + (vm, None) }; - (vm, Some(logger)) - } else { - let vm_db = StoreVmDatabase::new(self.storage.clone(), parent_header.clone())?; - let vm = self.new_evm(vm_db)?; - (vm, None) - }; let ( res, @@ -1936,18 +2131,33 @@ impl Blockchain { block.body.transactions.len(), ); + let mut produced_witness: Option = None; + if let Some(logger) = logger && let Some(account_updates) = accumulated_updates { - let block_hash = block.hash(); let witness = self.generate_witness_from_account_updates( account_updates, &block, parent_header, &logger, )?; - self.storage - .store_witness(block_hash, block_number, witness)?; + + if compute_witness { + // Convert to RPC format for returning over the wire. + let block_hash = block.hash(); + self.storage + .store_witness(block_hash, block_number, witness.clone())?; + + let rpc_witness = RpcExecutionWitness::try_from(witness) + .map_err(|e| ChainError::Custom(format!("witness conversion failed: {e}")))?; + produced_witness = Some(rpc_witness); + } else { + // Persist to DB for later retrieval via debug_executionWitness. + let block_hash = block.hash(); + self.storage + .store_witness(block_hash, block_number, witness)?; + } }; // Store the produced BAL (present on Amsterdam+ blocks) so peers can request it @@ -1981,7 +2191,6 @@ impl Blockchain { instants, ); } - metrics!(if let Some(bal_ref) = produced_bal.as_ref().or(bal) { let account_count = bal_ref.accounts().len() as u64; let slot_count = bal_ref.item_count().saturating_sub(account_count); @@ -1993,7 +2202,7 @@ impl Blockchain { METRICS_BAL.slot_count.set(slot_count as i64); }); - Ok((produced_bal, result)) + Ok((produced_bal, result.map(|()| produced_witness))) } #[allow(clippy::too_many_arguments)] diff --git a/crates/networking/rpc/Cargo.toml b/crates/networking/rpc/Cargo.toml index 27852b0ce01..811cc554041 100644 --- a/crates/networking/rpc/Cargo.toml +++ b/crates/networking/rpc/Cargo.toml @@ -42,6 +42,8 @@ jemalloc_pprof = { version = "0.8.0", optional = true, features = [ ] } spawned-rt.workspace = true spawned-concurrency.workspace = true +libssz = { workspace = true } +libssz-derive = { workspace = true } # EIP-8025 dependencies (optional) libssz-merkle = { workspace = true, optional = true } diff --git a/crates/networking/rpc/engine/mod.rs b/crates/networking/rpc/engine/mod.rs index 753cec1aab5..66944c003a7 100644 --- a/crates/networking/rpc/engine/mod.rs +++ b/crates/networking/rpc/engine/mod.rs @@ -3,6 +3,11 @@ pub mod client_version; pub mod exchange_transition_config; pub mod fork_choice; pub mod payload; +#[cfg(feature = "eip-8025")] +pub mod proof; +#[cfg(feature = "eip-8025")] +pub mod proof_types; +pub mod rest; use crate::{ rpc::{RpcApiContext, RpcHandler}, @@ -42,6 +47,17 @@ pub const CAPABILITIES: [&str; 24] = [ "engine_getClientVersionV1", ]; +/// REST capabilities advertised via `engine_exchangeCapabilities`. +pub const REST_CAPABILITIES: [&str; 1] = ["rest_engine_newPayloadWithWitness"]; + +/// EIP-8025 proof capabilities, advertised only when the feature is enabled. +#[cfg(feature = "eip-8025")] +pub const EIP8025_CAPABILITIES: [&str; 3] = [ + "engine_requestProofsV1", + "engine_verifyExecutionProofV1", + "engine_verifyNewPayloadRequestHeaderV1", +]; + impl From for RpcRequest { fn from(val: ExchangeCapabilitiesRequest) -> Self { RpcRequest { @@ -66,6 +82,10 @@ impl RpcHandler for ExchangeCapabilitiesRequest { } async fn handle(&self, _context: RpcApiContext) -> Result { - Ok(json!(CAPABILITIES)) + let mut caps: Vec<&str> = CAPABILITIES.to_vec(); + caps.extend_from_slice(&REST_CAPABILITIES); + #[cfg(feature = "eip-8025")] + caps.extend_from_slice(&EIP8025_CAPABILITIES); + Ok(json!(caps)) } } diff --git a/crates/networking/rpc/engine/payload.rs b/crates/networking/rpc/engine/payload.rs index 520d2bb8266..6a53910cbaa 100644 --- a/crates/networking/rpc/engine/payload.rs +++ b/crates/networking/rpc/engine/payload.rs @@ -890,7 +890,7 @@ fn validate_execution_payload_v2(payload: &ExecutionPayload) -> Result<(), RpcEr Ok(()) } -fn validate_execution_payload_v3(payload: &ExecutionPayload) -> Result<(), RpcErr> { +pub fn validate_execution_payload_v3(payload: &ExecutionPayload) -> Result<(), RpcErr> { // Validate that only the required arguments are present if payload.withdrawals.is_none() { return Err(RpcErr::WrongParam("withdrawals".to_string())); @@ -906,7 +906,7 @@ fn validate_execution_payload_v3(payload: &ExecutionPayload) -> Result<(), RpcEr } #[inline] -fn validate_execution_payload_v4(payload: &ExecutionPayload) -> Result<(), RpcErr> { +pub fn validate_execution_payload_v4(payload: &ExecutionPayload) -> Result<(), RpcErr> { // This method follows the same specification as `engine_newPayloadV4` additionally // rejects payload without block access list @@ -966,6 +966,44 @@ async fn validate_ancestors( Ok(None) } +// This function is used to make sure neither the current block nor its parent have been invalidated +// similar to validate_ancestors but returns PayloadStatusWithWitness +async fn validate_ancestors_with_witness( + block: &Block, + context: &RpcApiContext, +) -> Result, RpcErr> { + // Check if the block has already been invalidated + if let Some(latest_valid_hash) = context + .storage + .get_latest_valid_ancestor(block.hash()) + .await? + { + return Ok(Some(PayloadStatusWithWitness::invalid_with( + latest_valid_hash, + "Header has been previously invalidated.".into(), + ))); + } + + // Check if the parent block has already been invalidated + if let Some(latest_valid_hash) = context + .storage + .get_latest_valid_ancestor(block.header.parent_hash) + .await? + { + // Invalidate child too + context + .storage + .set_latest_valid_ancestor(block.header.hash(), latest_valid_hash) + .await?; + return Ok(Some(PayloadStatusWithWitness::invalid_with( + latest_valid_hash, + "Parent header has been previously invalidated.".into(), + ))); + } + + Ok(None) +} + async fn handle_new_payload_v1_v2( payload: &ExecutionPayload, block: Block, @@ -1000,7 +1038,42 @@ async fn handle_new_payload_v1_v2( Ok(payload_status) } -async fn handle_new_payload_v3( +async fn handle_new_payload_v1_v2_with_witness( + payload: &ExecutionPayload, + block: Block, + context: RpcApiContext, + bal: Option, +) -> Result { + let Some(syncer) = &context.syncer else { + return Err(RpcErr::Internal( + "New payload requested but syncer is not initialized".to_string(), + )); + }; + // Validate block hash + if let Err(RpcErr::Internal(error_msg)) = validate_block_hash(payload, &block) { + return Ok(PayloadStatusWithWitness::invalid_with_err(&error_msg)); + } + + // Check for invalid ancestors + if let Some(status) = validate_ancestors_with_witness(&block, &context).await? { + return Ok(status); + } + + // We have validated ancestors, the parent is correct + let latest_valid_hash = block.header.parent_hash; + + if syncer.sync_mode() == SyncMode::Snap { + debug!("Snap sync in progress, skipping new payload validation"); + return Ok(PayloadStatusWithWitness::syncing()); + } + + // All checks passed, execute payload + let payload_status = + try_execute_payload_with_witness(block, &context, latest_valid_hash, bal).await?; + Ok(payload_status) +} + +pub async fn handle_new_payload_v3( payload: &ExecutionPayload, context: RpcApiContext, block: Block, @@ -1024,6 +1097,30 @@ async fn handle_new_payload_v3( handle_new_payload_v1_v2(payload, block, context, bal).await } +pub async fn handle_new_payload_v3_with_witness( + payload: &ExecutionPayload, + context: RpcApiContext, + block: Block, + expected_blob_versioned_hashes: Vec, + bal: Option, +) -> Result { + // V3 specific: validate blob hashes + let blob_versioned_hashes: Vec = block + .body + .transactions + .iter() + .flat_map(|tx| tx.blob_versioned_hashes()) + .collect(); + + if expected_blob_versioned_hashes != blob_versioned_hashes { + return Ok(PayloadStatusWithWitness::invalid_with_err( + "Invalid blob_versioned_hashes", + )); + } + + handle_new_payload_v1_v2_with_witness(payload, block, context, bal).await +} + async fn handle_new_payload_v4( payload: &ExecutionPayload, context: RpcApiContext, @@ -1039,9 +1136,25 @@ async fn handle_new_payload_v4( handle_new_payload_v3(payload, context, block, expected_blob_versioned_hashes, bal).await } +pub async fn handle_new_payload_v4_with_witness( + payload: &ExecutionPayload, + context: RpcApiContext, + block: Block, + expected_blob_versioned_hashes: Vec, + bal: Option, +) -> Result { + if let Some(bal) = &bal + && let Err(err) = bal.validate_ordering() + { + return Ok(PayloadStatusWithWitness::invalid_with_err(&err)); + } + handle_new_payload_v3_with_witness(payload, context, block, expected_blob_versioned_hashes, bal) + .await +} + // Elements of the list MUST be ordered by request_type in ascending order. // Elements with empty request_data MUST be excluded from the list. -fn validate_execution_requests(execution_requests: &[EncodedRequests]) -> Result<(), RpcErr> { +pub fn validate_execution_requests(execution_requests: &[EncodedRequests]) -> Result<(), RpcErr> { let mut last_type: i32 = -1; for requests in execution_requests { if requests.0.len() < 2 { @@ -1056,7 +1169,7 @@ fn validate_execution_requests(execution_requests: &[EncodedRequests]) -> Result Ok(()) } -fn get_block_from_payload( +pub fn get_block_from_payload( payload: &ExecutionPayload, parent_beacon_block_root: Option, requests_hash: Option, @@ -1178,6 +1291,89 @@ async fn try_execute_payload( } } +async fn try_execute_payload_with_witness( + block: Block, + context: &RpcApiContext, + latest_valid_hash: H256, + bal: Option, +) -> Result { + let Some(syncer) = &context.syncer else { + return Err(RpcErr::Internal( + "New payload requested but syncer is not initialized".to_string(), + )); + }; + let block_hash = block.hash(); + let block_number = block.header.number; + let storage = &context.storage; + // Return the valid message directly if we already have it. + // We check for header only as we do not download the block bodies before the pivot during snap sync + // https://github.com/lambdaclass/ethrex/issues/1766 + if storage.get_block_header_by_hash(block_hash)?.is_some() { + return Ok(PayloadStatusWithWitness::valid_with_hash(block_hash)); + } + + // Execute and store the block + debug!(%block_hash, %block_number, "Executing payload"); + + match add_block_with_witness(context, block, bal).await { + Err(ChainError::ParentNotFound) => { + // Start sync + syncer.sync_to_head(block_hash); + Ok(PayloadStatusWithWitness::syncing()) + } + // Under the current implementation this is not possible: we always calculate the state + // transition of any new payload as long as the parent is present. If we received the + // parent payload but it was stashed, then new payload would stash this one too, with a + // ParentNotFoundError. + Err(ChainError::ParentStateNotFound) => { + let e = "Failed to obtain parent state"; + error!("{e} for block {block_hash}"); + Err(RpcErr::Internal(e.to_string())) + } + Err(ChainError::InvalidBlock(error)) => { + warn!("Error executing block: {error}"); + context + .storage + .set_latest_valid_ancestor(block_hash, latest_valid_hash) + .await?; + Ok(PayloadStatusWithWitness::invalid_with( + latest_valid_hash, + error.to_string(), + )) + } + Err(ChainError::EvmError(error)) => { + warn!("Error executing block: {error}"); + context + .storage + .set_latest_valid_ancestor(block_hash, latest_valid_hash) + .await?; + Ok(PayloadStatusWithWitness::invalid_with( + latest_valid_hash, + error.to_string(), + )) + } + Err(ChainError::StoreError(error)) => { + warn!("Error storing block: {error}"); + Err(RpcErr::Internal(error.to_string())) + } + Err(e) => { + error!("{e} for block {block_hash}"); + Err(RpcErr::Internal(e.to_string())) + } + Ok(witness) => { + debug!("Block with hash {block_hash} executed and added to storage successfully"); + let Some(witness) = witness else { + debug!("No witness found for block {block_hash}"); + return Ok(PayloadStatusWithWitness::valid_with_hash(block_hash)); + }; + + Ok(PayloadStatusWithWitness::valid_with_witness_and_hash( + witness, block_hash, + )) + } + } +} + fn parse_get_payload_request(params: &Option>) -> Result { let params = params .as_ref() @@ -1254,3 +1450,113 @@ async fn get_payload(payload_id: u64, context: &RpcApiContext) -> Result, + pub validation_error: Option, + pub witness: Option, +} + +impl PayloadStatusWithWitness { + // Convenience methods to create payload status + + pub fn invalid_with(latest_valid_hash: H256, error: String) -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Invalid, + latest_valid_hash: Some(latest_valid_hash), + validation_error: Some(error), + witness: None, + } + } + + /// Creates a PayloadStatus with invalid status and error message + pub fn invalid_with_err(error: &str) -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Invalid, + latest_valid_hash: None, + validation_error: Some(error.to_string()), + witness: None, + } + } + + /// Creates a PayloadStatus with invalid status and latest valid hash + pub fn invalid_with_hash(hash: BlockHash) -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Invalid, + latest_valid_hash: Some(hash), + validation_error: None, + witness: None, + } + } + + /// Creates a PayloadStatus with syncing status and no other info + pub fn syncing() -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Syncing, + latest_valid_hash: None, + validation_error: None, + witness: None, + } + } + + /// Creates a PayloadStatus with valid status and latest valid hash + pub fn valid_with_hash(hash: BlockHash) -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Valid, + latest_valid_hash: Some(hash), + validation_error: None, + witness: None, + } + } + /// Creates a PayloadStatus with valid status and latest valid hash + pub fn valid() -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Valid, + latest_valid_hash: None, + validation_error: None, + witness: None, + } + } + + pub fn valid_with_witness_and_hash(witness: RpcExecutionWitness, hash: BlockHash) -> Self { + PayloadStatusWithWitness { + status: PayloadValidationStatus::Valid, + latest_valid_hash: Some(hash), + validation_error: None, + witness: Some(witness), + } + } +} + +/// Channel message type for the witness block executor. +pub type WitnessBlockWorkerMessage = ( + oneshot::Sender, ChainError>>, + Block, + Option, +); + +/// Submit a block for execution + witness generation via the witness worker. +pub async fn add_block_with_witness( + ctx: &RpcApiContext, + block: Block, + bal: Option, +) -> Result, ChainError> { + let (notify_send, notify_recv) = oneshot::channel(); + ctx.witness_block_worker_channel + .as_ref() + .ok_or_else(|| ChainError::Custom("witness block worker not available".to_string()))? + .send((notify_send, block, bal)) + .map_err(|e| { + ChainError::Custom(format!( + "failed to send block execution request to worker: {e}" + )) + })?; + notify_recv + .await + .map_err(|e| ChainError::Custom(format!("failed to receive block execution result: {e}")))? +} diff --git a/crates/networking/rpc/engine/rest.rs b/crates/networking/rpc/engine/rest.rs new file mode 100644 index 00000000000..0057879a6bd --- /dev/null +++ b/crates/networking/rpc/engine/rest.rs @@ -0,0 +1,437 @@ +//! REST bindings for the Engine API. +//! The endpoint accepts the same JSON parameters as `engine_newPayloadV5`, +//! executes the block, generates a witness, and returns the result as +//! SSZ-encoded bytes (`Content-Type: application/octet-stream`). + +use axum::{ + Json, + body::Body, + extract::State, + http::{Response, StatusCode, header}, + response::IntoResponse, +}; +use axum_extra::{ + TypedHeader, + headers::{Authorization, authorization::Bearer}, +}; +use bytes::Bytes; +use libssz::SszEncode; +use libssz_derive::SszEncode; +use serde_json::Value; + +use ethrex_common::H256; +use ethrex_common::types::block_execution_witness::RpcExecutionWitness; +use ethrex_common::types::requests::{EncodedRequests, compute_requests_hash}; + +use crate::{authentication::authenticate, engine::payload::get_block_from_payload}; +use crate::{engine::payload::handle_new_payload_v3_with_witness, rpc::RpcApiContext}; +use crate::{engine::payload::validate_execution_payload_v3, utils::RpcErr}; +use crate::{ + engine::payload::{handle_new_payload_v4_with_witness, validate_execution_payload_v4}, + types::payload::{ExecutionPayload, PayloadValidationStatus}, +}; + +use super::payload; + +/// `Union[None, ByteVector[32]]` for `latest_valid_hash`. +#[derive(SszEncode)] +#[ssz(enum_behaviour = "union")] +pub enum OptionalHash { + None, + Hash([u8; 32]), +} + +/// `Union[None, List[uint8, VALIDATION_ERROR_MAX]]` for `validation_error`. +#[derive(SszEncode)] +#[ssz(enum_behaviour = "union")] +pub enum OptionalValidationError { + None, + Error(Vec), +} + +/// Maximum number of `uint8` elements for validation_error. +const VALIDATION_ERROR_MAX: usize = 8192; + +/// SSZ `Container` for the top-level response. +/// +/// ```text +/// status: uint8 +/// latest_valid_hash: Union[None, ByteVector[32]] +/// validation_error: Union[None, List[uint8, VALIDATION_ERROR_MAX]] +/// witness: List[uint8, MAX_WITNESS_BYTES] +/// ``` +#[derive(SszEncode)] +pub struct SszNewPayloadWithWitnessResponse { + pub status: u8, + pub latest_valid_hash: OptionalHash, + pub validation_error: OptionalValidationError, + pub witness: Vec, +} + +/// SSZ `Container` for `ExecutionWitnessV1`. +/// +/// ```text +/// state: List[List[uint8, MAX_WITNESS_ITEM_BYTES], MAX_WITNESS_ITEMS] +/// keys: List[List[uint8, MAX_WITNESS_ITEM_BYTES], MAX_WITNESS_ITEMS] +/// codes: List[List[uint8, MAX_WITNESS_ITEM_BYTES], MAX_WITNESS_ITEMS] +/// headers: List[List[uint8, MAX_WITNESS_ITEM_BYTES], MAX_WITNESS_ITEMS] +/// ``` +#[derive(SszEncode)] +pub struct SszExecutionWitnessV1 { + pub state: Vec>, + pub keys: Vec>, + pub codes: Vec>, + pub headers: Vec>, +} + +impl SszNewPayloadWithWitnessResponse { + /// Build from a validation status + optional witness. + fn from_status( + status: PayloadValidationStatus, + latest_valid_hash: Option, + validation_error: Option, + witness: Option, + ) -> Self { + let status_byte = match status { + PayloadValidationStatus::Valid => 0u8, + PayloadValidationStatus::Invalid => 1u8, + PayloadValidationStatus::Syncing => 2u8, + PayloadValidationStatus::Accepted => 3u8, + }; + + let ssz_hash = match latest_valid_hash { + Some(h) => OptionalHash::Hash(h.0), + None => OptionalHash::None, + }; + + let ssz_error = match validation_error { + Some(msg) => { + let mut bytes = msg.into_bytes(); + if bytes.len() > VALIDATION_ERROR_MAX { + let mut end = VALIDATION_ERROR_MAX; + while end > 0 && (bytes[end] & 0xC0) == 0x80 { + end -= 1; + } + bytes.truncate(end); + } + OptionalValidationError::Error(bytes) + } + None => OptionalValidationError::None, + }; + + let witness_bytes = if status_byte == 0 { + match witness { + Some(w) => { + let ssz_witness = SszExecutionWitnessV1::from(w); + ssz_witness.to_ssz() + } + None => Vec::new(), + } + } else { + Vec::new() + }; + + Self { + status: status_byte, + latest_valid_hash: ssz_hash, + validation_error: ssz_error, + witness: witness_bytes, + } + } +} + +impl From for SszExecutionWitnessV1 { + fn from(w: RpcExecutionWitness) -> Self { + Self { + state: w.state.into_iter().map(|b| b.to_vec()).collect(), + keys: w.keys.into_iter().map(|b| b.to_vec()).collect(), + codes: w.codes.into_iter().map(|b| b.to_vec()).collect(), + headers: w.headers.into_iter().map(|b| b.to_vec()).collect(), + } + } +} + +/// JSON error body matching the JSON-RPC `error` shape. +fn json_error_response(code: i32, message: &str) -> axum::response::Response { + let status = match code { + -32700 | -32602 => StatusCode::BAD_REQUEST, + -38005 => StatusCode::BAD_REQUEST, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + let body = serde_json::json!({ + "code": code, + "message": message, + }); + (status, Json(body)).into_response() +} + +fn rpc_err_to_response(err: RpcErr) -> axum::response::Response { + match &err { + RpcErr::BadParams(msg) | RpcErr::WrongParam(msg) | RpcErr::InvalidPayload(msg) => { + json_error_response(-32602, msg) + } + RpcErr::UnsupportedFork(msg) => json_error_response(-38005, msg), + RpcErr::Internal(msg) => json_error_response(-32603, msg), + _ => json_error_response(-32603, &format!("{err:?}")), + } +} + +/// `POST /new-payload-with-witness-v4` +pub async fn handle_new_payload_with_witness_v4( + State(context): State, + auth_header: Option>>, + body: axum::body::Bytes, +) -> axum::response::Response { + if let Err(_err) = authenticate(&context.node_data.jwt_secret, auth_header) { + return json_error_response(-32700, "JWT authentication failed"); + } + + let params: Vec = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(e) => { + return json_error_response(-32700, &format!("Parse error: {e}")); + } + }; + + let ( + exec_payload, + expected_blob_versioned_hashes, + parent_beacon_block_root, + execution_requests, + ) = match parse_params(params) { + Ok(res) => res, + Err(e) => return json_error_response(-32602, &format!("Parse error: {e}")), + }; + + if let Err(e) = payload::validate_execution_requests(&execution_requests) { + return rpc_err_to_response(e); + } + + let requests_hash = compute_requests_hash(&execution_requests); + + let block = match get_block_from_payload( + &exec_payload, + Some(parent_beacon_block_root), + Some(requests_hash), + None, + ) { + Ok(block) => block, + Err(err) => { + let resp = SszNewPayloadWithWitnessResponse::from_status( + PayloadValidationStatus::Invalid, + None, + Some(err.to_string()), + None, + ); + return ssz_response(resp); + } + }; + + let chain_config = context.storage.get_chain_config(); + + if !chain_config.is_prague_activated(block.header.timestamp) { + return rpc_err_to_response(RpcErr::UnsupportedFork(format!( + "{:?}", + chain_config.get_fork(block.header.timestamp) + ))); + } + // We use v3 since the execution payload remains the same. + if let Err(e) = validate_execution_payload_v3(&exec_payload) { + return rpc_err_to_response(e); + } + let payload_result = handle_new_payload_v3_with_witness( + &exec_payload, + context, + block, + expected_blob_versioned_hashes, + None, + ) + .await; + + match payload_result { + Ok(payload_status) => { + let resp = SszNewPayloadWithWitnessResponse::from_status( + payload_status.status, + payload_status.latest_valid_hash, + payload_status.validation_error, + payload_status.witness, + ); + ssz_response(resp) + } + Err(rpc_err) => rpc_err_to_response(rpc_err), + } +} + +/// `POST /new-payload-with-witness-v5` +pub async fn handle_new_payload_with_witness_v5( + State(context): State, + auth_header: Option>>, + body: axum::body::Bytes, +) -> axum::response::Response { + if let Err(_err) = authenticate(&context.node_data.jwt_secret, auth_header) { + return json_error_response(-32700, "JWT authentication failed"); + } + + let params: Vec = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(e) => { + return json_error_response(-32700, &format!("Parse error: {e}")); + } + }; + + // Guard against empty / short params before any direct indexing. + if params.is_empty() { + return json_error_response(-32602, "Parse error: Expected 4 params, got 0"); + } + + // Extract the raw BAL hash from the JSON payload before deserialization. + // We hash the raw RLP bytes as-received to preserve the exact encoding + // (including any ordering) for accurate block hash validation. + let Ok(raw_bal_hash) = params[0] + .get("blockAccessList") + .map(|v| { + let hex_str = v + .as_str() + .ok_or(RpcErr::WrongParam("blockAccessList".to_string()))?; + let bytes = hex::decode(hex_str.trim_start_matches("0x")) + .map_err(|_| RpcErr::WrongParam("blockAccessList".to_string()))?; + Ok::<_, RpcErr>(ethrex_common::utils::keccak(bytes)) + }) + .transpose() + else { + return json_error_response(-32602, "Invalid blockAccessList"); + }; + + let ( + exec_payload, + expected_blob_versioned_hashes, + parent_beacon_block_root, + execution_requests, + ) = match parse_params(params) { + Ok(res) => res, + Err(e) => return json_error_response(-32602, &format!("Parse error: {e}")), + }; + + if let Err(e) = validate_execution_payload_v4(&exec_payload) { + return rpc_err_to_response(e); + } + + if let Err(e) = payload::validate_execution_requests(&execution_requests) { + return rpc_err_to_response(e); + } + + let requests_hash = compute_requests_hash(&execution_requests); + // Use the hash computed from the raw RLP bytes as-received. + // This preserves the exact encoding (including any ordering) from the payload, + // so the block hash check correctly detects BAL corruption. + let block_access_list_hash = raw_bal_hash; + + let block = match get_block_from_payload( + &exec_payload, + Some(parent_beacon_block_root), + Some(requests_hash), + block_access_list_hash, + ) { + Ok(block) => block, + Err(err) => { + let resp = SszNewPayloadWithWitnessResponse::from_status( + PayloadValidationStatus::Invalid, + None, + Some(err.to_string()), + None, + ); + return ssz_response(resp); + } + }; + + let chain_config = context.storage.get_chain_config(); + + if !chain_config.is_amsterdam_activated(block.header.timestamp) { + return rpc_err_to_response(RpcErr::UnsupportedFork(format!( + "{:?}", + chain_config.get_fork(block.header.timestamp) + ))); + } + + let bal = exec_payload.block_access_list.clone(); + let payload_result = handle_new_payload_v4_with_witness( + &exec_payload, + context, + block, + expected_blob_versioned_hashes, + bal, + ) + .await; + + match payload_result { + Ok(payload_status) => { + let resp = SszNewPayloadWithWitnessResponse::from_status( + payload_status.status, + payload_status.latest_valid_hash, + payload_status.validation_error, + payload_status.witness, + ); + ssz_response(resp) + } + Err(rpc_err) => rpc_err_to_response(rpc_err), + } +} + +/// Produce a `200 OK` response with `Content-Type: application/octet-stream`. +fn ssz_response(resp: SszNewPayloadWithWitnessResponse) -> axum::response::Response { + let mut buf = Vec::with_capacity(resp.encoded_len()); + resp.ssz_append(&mut buf); + let bytes = Bytes::from(buf); + + Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/octet-stream") + .body(Body::from(bytes)) + .expect("failed to build SSZ response") + .into_response() +} + +/// Parse the parameters for the new payload with witness RPC methods. +fn parse_params( + params: Vec, +) -> Result<(ExecutionPayload, Vec, H256, Vec), String> { + if params.len() != 4 { + return Err(format!("Expected 4 params, got {}", params.len())); + } + + let exec_payload: ExecutionPayload = match serde_json::from_value(params[0].clone()) { + Ok(p) => p, + Err(_) => { + return Err(format!("Invalid executionPayload")); + } + }; + + let expected_blob_versioned_hashes: Vec = match serde_json::from_value(params[1].clone()) + { + Ok(h) => h, + Err(_) => { + return Err(format!("Invalid expectedBlobVersionedHashes")); + } + }; + + let parent_beacon_block_root: H256 = match serde_json::from_value(params[2].clone()) { + Ok(r) => r, + Err(_) => { + return Err(format!("Invalid parentBeaconBlockRoot")); + } + }; + + let execution_requests: Vec = match serde_json::from_value(params[3].clone()) { + Ok(r) => r, + Err(_) => { + return Err(format!("Invalid executionRequests")); + } + }; + + Ok(( + exec_payload, + expected_blob_versioned_hashes, + parent_beacon_block_root, + execution_requests, + )) +} diff --git a/crates/networking/rpc/rpc.rs b/crates/networking/rpc/rpc.rs index e9687c3c789..7c14922570f 100644 --- a/crates/networking/rpc/rpc.rs +++ b/crates/networking/rpc/rpc.rs @@ -4,7 +4,14 @@ use crate::debug::execution_witness::ExecutionWitnessRequest; use crate::debug::execution_witness_by_hash::ExecutionWitnessByBlockHashRequest; use crate::engine::blobs::{BlobsV2Request, BlobsV3Request}; use crate::engine::client_version::GetClientVersionV1Request; -use crate::engine::payload::{GetPayloadV5Request, GetPayloadV6Request, NewPayloadV5Request}; +use crate::engine::payload::{ + GetPayloadV5Request, GetPayloadV6Request, NewPayloadV5Request, WitnessBlockWorkerMessage, +}; +#[cfg(feature = "eip-8025")] +use crate::engine::proof::{ + RequestProofsV1, VerifyExecutionProofV1, VerifyNewPayloadRequestHeaderV1, +}; +use crate::engine::rest::{handle_new_payload_with_witness_v4, handle_new_payload_with_witness_v5}; use crate::engine::{ ExchangeCapabilitiesRequest, blobs::BlobsV1Request, @@ -219,6 +226,12 @@ pub struct RpcApiContext { /// The `engine` namespace is always served via the authenticated RPC port /// and is not gated here. pub allowed_namespaces: Arc>, + /// Optional channel for the witness block executor (REST endpoint). + pub witness_block_worker_channel: Option>, + /// EIP-8025 proof coordinator handle for sending proof requests. + #[cfg(feature = "eip-8025")] + pub proof_coordinator: + Option, } /// Configuration for the WebSocket RPC server. @@ -456,6 +469,28 @@ pub fn start_block_executor(blockchain: Arc) -> UnboundedSender, +) -> UnboundedSender { + let (tx, mut rx) = unbounded_channel::(); + std::thread::Builder::new() + .name("witness_block_executor".to_string()) + .spawn(move || { + while let Some((notify, block, bal)) = rx.blocking_recv() { + let _ = notify + .send(blockchain.add_block_pipeline_with_witness(block, bal.as_ref())) + .inspect_err(|_| tracing::error!("failed to notify witness caller")); + } + }) + .expect("Failed to spawn witness_block_executor thread"); + tx +} + /// Starts the JSON-RPC API servers. /// /// This function initializes and runs up to three server endpoints: @@ -518,6 +553,7 @@ pub async fn start_api( // filters are used by the filters endpoints (eth_newFilter, eth_getFilterChanges, ...etc) let active_filters = Arc::new(Mutex::new(HashMap::new())); let block_worker_channel = start_block_executor(blockchain.clone()); + let witness_block_worker_channel = start_witness_block_executor(blockchain.clone()); let service_context = RpcApiContext { storage, blockchain, @@ -537,6 +573,9 @@ pub async fn start_api( block_worker_channel, ws: ws.clone(), allowed_namespaces: Arc::new(allowed_namespaces), + witness_block_worker_channel: Some(witness_block_worker_channel), + #[cfg(feature = "eip-8025")] + proof_coordinator, }; // Periodically clean up the active filters for the filters endpoints. @@ -592,6 +631,14 @@ pub async fn start_api( let authrpc_router = Router::new() .route("/", post(authrpc_handler)) + .route( + "/new-payload-with-witness-v4", + post(handle_new_payload_with_witness_v4), + ) + .route( + "/new-payload-with-witness", + post(handle_new_payload_with_witness_v5), + ) .with_state(service_context.clone()) // Bump the body limit for the engine API to 256MB // This is needed to receive payloads bigger than the default limit of 2MB diff --git a/crates/networking/rpc/test_utils.rs b/crates/networking/rpc/test_utils.rs index 4906b91f857..7148a052f00 100644 --- a/crates/networking/rpc/test_utils.rs +++ b/crates/networking/rpc/test_utils.rs @@ -317,6 +317,9 @@ pub async fn default_context_with_storage(storage: Store) -> RpcApiContext { block_worker_channel, ws: None, allowed_namespaces: Arc::new(all_namespaces_for_tests()), + witness_block_worker_channel: None, + #[cfg(feature = "eip-8025")] + proof_coordinator: None, } } diff --git a/fixtures/networks/default.yaml b/fixtures/networks/default.yaml index fbf2b749ec9..c3de8c46692 100644 --- a/fixtures/networks/default.yaml +++ b/fixtures/networks/default.yaml @@ -49,4 +49,4 @@ spamoor_params: throughput: 750 grafana_params: - additional_dashboards: ["./ethrex_l1_perf.json"] + additional_dashboards: ["./ethrex_l1_perf.json"] \ No newline at end of file