diff --git a/cmd/ethrex/cli.rs b/cmd/ethrex/cli.rs index feb194d7cfa..c1e6a344a97 100644 --- a/cmd/ethrex/cli.rs +++ b/cmd/ethrex/cli.rs @@ -996,9 +996,10 @@ pub async fn import_blocks( block_batch.push(block); if block_batch.len() >= IMPORT_BATCH_SIZE || index + MIN_FULL_BLOCKS + 1 == size { blockchain + .clone() .add_blocks_in_batch( mem::take(&mut block_batch), - &[], + vec![], CancellationToken::new(), ) .await diff --git a/crates/blockchain/blockchain.rs b/crates/blockchain/blockchain.rs index 210206dea1d..7b92c514f41 100644 --- a/crates/blockchain/blockchain.rs +++ b/crates/blockchain/blockchain.rs @@ -2881,12 +2881,29 @@ impl Blockchain { /// the failed block hash and the last successfully-imported block hash. /// /// `bals` holds the per-block Block Access Lists fetched during sync, aligned by index with - /// `blocks`. Pass an empty slice when no BALs are available (e.g. block import from RLP); the + /// `blocks`. Pass an empty vec when no BALs are available (e.g. block import from RLP); the /// pipeline then rebuilds each BAL. Only a BAL matching its block's header commitment is used. + /// + /// The per-block pipeline is CPU/DB-bound, so the loop runs on Tokio's blocking pool and + /// does not hold an async worker for the duration of each block. pub async fn add_blocks_in_batch( + self: Arc, + blocks: Vec, + bals: Vec>, + cancellation_token: CancellationToken, + ) -> Result<(), (ChainError, Option)> { + tokio::task::spawn_blocking(move || { + self.add_blocks_in_batch_sync(blocks, bals, cancellation_token) + }) + .await + .map_err(|e| (ChainError::Custom(e.to_string()), None))? + } + + /// Blocking loop for [`Self::add_blocks_in_batch`]. + fn add_blocks_in_batch_sync( &self, blocks: Vec, - bals: &[Option], + bals: Vec>, cancellation_token: CancellationToken, ) -> Result<(), (ChainError, Option)> { debug_assert!( @@ -2949,7 +2966,6 @@ impl Blockchain { transactions_count += block_tx_count; log_batch_progress(blocks_len as u32, i as u32); - tokio::task::yield_now().await; } let elapsed_seconds = interval.elapsed().as_secs_f64(); diff --git a/crates/networking/p2p/sync/full.rs b/crates/networking/p2p/sync/full.rs index ca3b3f81f04..f29d03879dd 100644 --- a/crates/networking/p2p/sync/full.rs +++ b/crates/networking/p2p/sync/full.rs @@ -12,7 +12,7 @@ use ethrex_common::{ H256, types::{Block, BlockBody, BlockHeader, block_access_list::BlockAccessList}, }; -use ethrex_storage::{DB_COMMIT_THRESHOLD, Store}; +use ethrex_storage::Store; use tokio::sync::RwLock; use tokio::time::Instant; use tokio_util::sync::CancellationToken; @@ -195,7 +195,6 @@ pub async fn sync_cycle_full( blockchain.clone(), cancel_token.clone(), pending_blocks, - true, store.clone(), peers, ) @@ -538,7 +537,6 @@ pub async fn sync_cycle_full( blockchain.clone(), cancel_token.clone(), blocks, - final_batch, store.clone(), peers, ) @@ -581,7 +579,6 @@ pub async fn sync_cycle_full( blockchain.clone(), cancel_token.clone(), pending_blocks, - true, store.clone(), peers, ) @@ -615,7 +612,6 @@ async fn add_blocks_in_batch( blockchain: Arc, cancel_token: CancellationToken, blocks: Vec, - final_batch: bool, store: Store, peers: &mut PeerHandler, ) -> Result<(), SyncError> { @@ -638,8 +634,8 @@ async fn add_blocks_in_batch( let blocks_hashes = blocks.iter().map(|block| block.hash()).collect::>(); let chain_config = store.get_chain_config(); let bals: Vec> = { - // Fetch BALs for every Amsterdam batch (not just the final one): both the - // batch path and `run_blocks_pipeline` now persist them, so peers can serve + // Fetch BALs for every Amsterdam batch (not just the final one): + // `add_blocks_in_batch` persists them, so peers can serve // these blocks over eth/71 later without regenerating against pruned state. let any_amsterdam = blocks .iter() @@ -658,7 +654,7 @@ async fn add_blocks_in_batch( }; // Run the batch if let Err((err, batch_failure)) = - add_blocks(blockchain.clone(), blocks, bals, final_batch, cancel_token).await + add_blocks(blockchain.clone(), blocks, bals, cancel_token).await { if let Some(batch_failure) = batch_failure { warn!("Failed to add block during FullSync: {err}"); @@ -709,52 +705,14 @@ async fn add_blocks_in_batch( Ok(()) } -/// Executes the given blocks and stores them. -/// -/// Both paths execute block-by-block through the same validated pipeline -/// (`add_block_pipeline_bounded`), which builds fresh per-block VM state. When the sync -/// head is found the blocks run sequentially on a blocking thread; otherwise -/// `add_blocks_in_batch` runs them with BAL fetching, progress logging and cancellation. +/// Execute and store `blocks` via [`Blockchain::add_blocks_in_batch`]. async fn add_blocks( blockchain: Arc, blocks: Vec, bals: Vec>, - sync_head_found: bool, cancel_token: CancellationToken, ) -> Result<(), (ChainError, Option)> { - if sync_head_found { - return run_blocks_pipeline(blockchain, blocks, bals).await; - } blockchain - .add_blocks_in_batch(blocks, &bals, cancel_token) + .add_blocks_in_batch(blocks, bals, cancel_token) .await } - -async fn run_blocks_pipeline( - blockchain: Arc, - blocks: Vec, - bals: Vec>, -) -> Result<(), (ChainError, Option)> { - tokio::task::spawn_blocking(move || { - let mut last_valid_hash = H256::default(); - for (block, bal) in blocks.into_iter().zip(bals.into_iter()) { - let block_hash = block.hash(); - blockchain - .add_block_pipeline_bounded(block, bal.map(Arc::new), DB_COMMIT_THRESHOLD) - .map(|_| ()) - .map_err(|e| { - ( - e, - Some(BatchBlockProcessingFailure { - last_valid_hash, - failed_block_hash: block_hash, - }), - ) - })?; - last_valid_hash = block_hash; - } - Ok(()) - }) - .await - .map_err(|e| (ChainError::Custom(e.to_string()), None))? -} diff --git a/test/tests/blockchain/batch_tests.rs b/test/tests/blockchain/batch_tests.rs index 270134c8a46..5cb1b68df28 100644 --- a/test/tests/blockchain/batch_tests.rs +++ b/test/tests/blockchain/batch_tests.rs @@ -1,4 +1,4 @@ -use std::{fs::File, io::BufReader, path::PathBuf}; +use std::{fs::File, io::BufReader, path::PathBuf, sync::Arc}; use bytes::Bytes; use ethrex_blockchain::{ @@ -234,10 +234,11 @@ async fn batch_selfdestruct_created_account_no_spurious_state() { // 7. Create a fresh store B and re-execute both blocks in batch. // This is the code path that was buggy before the fix. let (store_b, _) = setup_store(sender).await; - let blockchain_b = Blockchain::default_with_store(store_b); + let blockchain_b = Arc::new(Blockchain::default_with_store(store_b)); let result = blockchain_b - .add_blocks_in_batch(vec![block1, block2], &[], CancellationToken::new()) + .clone() + .add_blocks_in_batch(vec![block1, block2], vec![], CancellationToken::new()) .await; assert!( @@ -382,10 +383,11 @@ async fn batch_cross_batch_blockhash_regression() { // Batch 1: blocks 1-2, Batch 2: block 3 // Block 3 needs blockhash(2) which is only in batch 1. let (store_b, _) = setup_store(sender).await; - let blockchain_b = Blockchain::default_with_store(store_b); + let blockchain_b = Arc::new(Blockchain::default_with_store(store_b)); let result1 = blockchain_b - .add_blocks_in_batch(vec![block1, block2], &[], CancellationToken::new()) + .clone() + .add_blocks_in_batch(vec![block1, block2], vec![], CancellationToken::new()) .await; assert!( result1.is_ok(), @@ -394,7 +396,7 @@ async fn batch_cross_batch_blockhash_regression() { ); let result2 = blockchain_b - .add_blocks_in_batch(vec![block3], &[], CancellationToken::new()) + .add_blocks_in_batch(vec![block3], vec![], CancellationToken::new()) .await; assert!( result2.is_ok(), @@ -438,10 +440,10 @@ async fn batch_single_block_selfdestruct() { // Batch path: re-execute the same block on a fresh store. let (store_b, _) = setup_store(sender).await; - let blockchain_b = Blockchain::default_with_store(store_b); + let blockchain_b = Arc::new(Blockchain::default_with_store(store_b)); let result = blockchain_b - .add_blocks_in_batch(vec![block1], &[], CancellationToken::new()) + .add_blocks_in_batch(vec![block1], vec![], CancellationToken::new()) .await; assert!( diff --git a/test/tests/blockchain/canonical_commit_gate_tests.rs b/test/tests/blockchain/canonical_commit_gate_tests.rs index 2bb0d0005be..207df1f7995 100644 --- a/test/tests/blockchain/canonical_commit_gate_tests.rs +++ b/test/tests/blockchain/canonical_commit_gate_tests.rs @@ -368,7 +368,7 @@ async fn bounded_reexec_without_fcu_bounds_memory_and_serves_window() { ); // Re-execute a single canonical chain through the BOUNDED path, block by block, with - // NO forkchoice_update — exactly what regenerate_head_state / run_blocks_pipeline / + // NO forkchoice_update — exactly what regenerate_head_state / full-sync batch import / // import do. Record each block's state root to query the retained window later. let mut parent_header = genesis_header; let mut roots: Vec = Vec::with_capacity(BLOCKS as usize); // roots[N-1] == block N