perf(l1): thread Arc<BlockAccessList> to avoid per-block BAL deep clones (#6829)

Before this change, `execute_block_parallel` in
`crates/vm/backends/levm/mod.rs` called `Arc::new(bal.clone())` and
`Arc::new(validation_index.clone())` on the critical path -- a full deep
copy of the BAL (potentially multiple MB of account/slot/code data)
before a single transaction executed, serially.

Fix: thread `Option<Arc<BlockAccessList>>` and `Arc<BalAddressIndex>`
through the entire pipeline:
`add_block_pipeline{,_bal,_with_witness,_inner}`,
`Blockchain::execute_block_pipeline`, `Evm::execute_block_pipeline`,
`LEVM::execute_block_pipeline`, and `execute_block_parallel`. The owned
BAL is wrapped in an `Arc` once at the call boundary
(`bal.map(Arc::new)`). The validation index is built directly into an
`Arc`. All downstream per-thread (warmer/executor) and per-tx
(`LazyBalCursor`) uses become pointer clones. The two deep clones become
moves.

7 files touched: `cmd/ethrex/cli.rs`, `crates/blockchain/blockchain.rs`,
`crates/networking/p2p/sync/full.rs`, `crates/networking/rpc/rpc.rs`,
`crates/vm/backends/levm/mod.rs`, `crates/vm/backends/mod.rs`,
`tooling/ef_tests/blockchain/test_runner.rs`. Zero behavior change.

`make test` passes (57 binaries, 0 failures). `cargo check --workspace
--all-targets` and `cargo clippy` clean. Full EF state/blockchain suites
not run locally; they should gate in CI.

Author: edg-l

CHANGELOG.md

@@ -12,6 +12,10 @@
 
 ## Perf
 
+### 2026-06-29
+
+- Thread `Arc<BlockAccessList>` through the block pipeline to avoid an O(BAL-size) deep clone of the Block Access List (and its validation index) per block on the parallel execution path [#6829](https://github.com/lambdaclass/ethrex/pull/6829)
+
 ### 2026-06-18
 
 - In-place top-slot mutation for unary/binary opcodes and `MLOAD`: mutate the top stack slot directly instead of pop-then-push, removing the serial read-modify-write of the stack offset on offset-chain-bound ops. ~2.16x on an ISZERO loop and ~1.63x on `MLOAD` (IPC 2.41 -> 3.45 on a 33M-op loop) [#6865](https://github.com/lambdaclass/ethrex/pull/6865)

cmd/ethrex/cli.rs

@@ -1030,14 +1030,15 @@ pub async fn import_blocks_bench(
         info!(path = %bal_path, "Loading BALs from file (parallel path)");
         use ethrex_common::types::block_access_list::BlockAccessList;
         use ethrex_rlp::decode::RLPDecode as _;
+        use std::sync::Arc;
         let data = std::fs::read(bal_path)
             .unwrap_or_else(|e| panic!("failed to read BAL file at {bal_path:?}: {e}"));
         let mut remaining = data.as_slice();
-        let mut bals = Vec::new();
+        let mut bals: Vec<Arc<BlockAccessList>> = Vec::new();
         while !remaining.is_empty() {
             let (bal, rest) = BlockAccessList::decode_unfinished(remaining)
                 .unwrap_or_else(|e| panic!("failed to decode BAL from {bal_path:?}: {e}"));
-            bals.push(bal);
+            bals.push(Arc::new(bal));
             remaining = rest;
         }
         let amsterdam_blocks = chains
@@ -1104,7 +1105,10 @@ pub async fn import_blocks_bench(
             // BALs are only produced for Amsterdam+ blocks, so use a separate counter
             // that only advances for blocks that have a BAL hash in the header.
             let bal = if block.header.block_access_list_hash.is_some() {
-                let b = preloaded_bals.as_ref().and_then(|bals| bals.get(bal_index));
+                let b = preloaded_bals
+                    .as_ref()
+                    .and_then(|bals| bals.get(bal_index))
+                    .cloned();
                 bal_index += 1;
                 b
             } else {

crates/blockchain/blockchain.rs

@@ -522,7 +522,7 @@ impl Blockchain {
         block: &Block,
         parent_header: &BlockHeader,
         vm: &mut Evm,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
         collect_witness: bool,
     ) -> Result<BlockExecutionPipelineResult, ChainError> {
         let start_instant = Instant::now();
@@ -569,7 +569,7 @@ impl Blockchain {
         // consumes — the synthesized path would leave the receiver dropped.
         let optimistic_updates: Option<FxHashMap<Address, BalSynthesisItem>> =
             if self.options.bal_parallel_trie_enabled && !collect_witness {
-                bal.map(synthesize_bal_updates)
+                bal.as_deref().map(synthesize_bal_updates)
             } else {
                 None
             };
@@ -611,14 +611,18 @@ impl Blockchain {
         #[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
         if self.options.bal_prefetch_enabled
             && !collect_witness
-            && let Some(bal) = bal
+            && let Some(bal_ref) = bal.as_ref()
         {
-            let slots = LEVM::bal_storage_slots(bal);
+            let slots = LEVM::bal_storage_slots(bal_ref);
             if !slots.is_empty() {
                 let _ = caching_store.prefetch_storage(&slots);
             }
         }
 
+        // Each thread that captures `bal` needs its own Arc clone (cheap pointer bump).
+        #[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
+        let bal_warmer = bal.clone();
+
         let (execution_result, merkleization_result, warmer_duration) =
             std::thread::scope(|s| -> Result<_, ChainError> {
                 #[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
@@ -635,11 +639,11 @@ impl Blockchain {
                                 // Warming uses the same caching store, sharing cached state with execution.
                                 // Precompile cache lives inside CachingDatabase, shared automatically.
                                 let start = Instant::now();
-                                if let Some(bal) = bal {
+                                if let Some(bal) = bal_warmer {
                                     if bal_prefetch_enabled {
                                         // Amsterdam+: BAL-based precise prefetching (no tx re-execution).
                                         if let Err(e) = LEVM::warm_block_from_bal(
-                                            bal,
+                                            &bal,
                                             caching_store,
                                             cancelled_ref,
                                         ) {
@@ -701,6 +705,10 @@ impl Blockchain {
                 let execution_handle = std::thread::Builder::new()
                     .name("block_executor_execution".to_string())
                     .spawn_scoped(s, move || -> Result<_, ChainError> {
+                        // Cheap Arc pointer bump: `execute_block_pipeline` takes
+                        // ownership, but the header-commitment check below still needs
+                        // the input BAL on the parallel path (produced_bal == None).
+                        let header_bal = bal.clone();
                         let result = vm.execute_block_pipeline(
                             block,
                             tx,
@@ -761,7 +769,7 @@ impl Blockchain {
                                 block.body.transactions.len(),
                                 &NativeCrypto,
                             )?;
-                        } else if let Some(header_bal) = bal
+                        } else if let Some(header_bal) = header_bal.as_deref()
                             && chain_config.is_amsterdam_activated(block.header.timestamp)
                             && !header_bal.matches_commitment(
                                 block.header.block_access_list_hash,
@@ -2051,7 +2059,7 @@ impl Blockchain {
     pub fn add_block_pipeline(
         &self,
         block: Block,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
     ) -> Result<(), ChainError> {
         let (_, _, result) = self.add_block_pipeline_inner(block, bal, false)?;
         result
@@ -2066,7 +2074,7 @@ impl Blockchain {
     pub fn add_block_pipeline_bal(
         &self,
         block: Block,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
     ) -> Result<Option<BlockAccessList>, ChainError> {
         let (produced_bal, _, result) = self.add_block_pipeline_inner(block, bal, false)?;
         result?;
@@ -2078,7 +2086,7 @@ impl Blockchain {
     pub fn add_block_pipeline_with_witness(
         &self,
         block: Block,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
     ) -> Result<ExecutionWitness, ChainError> {
         let (_, witness, result) = self.add_block_pipeline_inner(block, bal, true)?;
         result?;
@@ -2100,7 +2108,7 @@ impl Blockchain {
     fn add_block_pipeline_inner(
         &self,
         block: Block,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
         force_witness: bool,
     ) -> Result<AddBlockPipelineInnerResult, ChainError> {
         // Validate if it can be the new head and find the parent
@@ -2143,6 +2151,9 @@ impl Blockchain {
             (vm, None)
         };
 
+        // Keep a copy of the input BAL Arc for the post-execution BAL store call below.
+        // `execute_block_pipeline` takes ownership; this clone is a cheap pointer bump.
+        let input_bal = bal.clone();
         let (
             res,
             account_updates_list,
@@ -2193,7 +2204,7 @@ impl Blockchain {
         // On the parallel Amsterdam validation path the BAL is supplied via the header
         // and `produced_bal` is None, so fall back to the validated incoming `bal`.
         // Pre-Amsterdam blocks have no BAL on either source, so nothing is stored.
-        if let Some(bal) = produced_bal.as_ref().or(bal)
+        if let Some(bal) = produced_bal.as_ref().or(input_bal.as_deref())
             && let Err(err) = self.storage.store_block_access_list(block_hash, bal)
         {
             warn!("Failed to store block access list for block {block_hash}: {err}");
@@ -2224,16 +2235,18 @@ impl Blockchain {
             );
         }
 
-        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);
-            let size_bytes = bal_ref.length() as f64;
-            METRICS_BAL.blocks_total.inc();
-            METRICS_BAL.size_bytes.set(size_bytes);
-            METRICS_BAL.size_bytes_histogram.observe(size_bytes);
-            METRICS_BAL.account_count.set(account_count as i64);
-            METRICS_BAL.slot_count.set(slot_count as i64);
-        });
+        metrics!(
+            if let Some(bal_ref) = produced_bal.as_ref().or(input_bal.as_deref()) {
+                let account_count = bal_ref.accounts().len() as u64;
+                let slot_count = bal_ref.item_count().saturating_sub(account_count);
+                let size_bytes = bal_ref.length() as f64;
+                METRICS_BAL.blocks_total.inc();
+                METRICS_BAL.size_bytes.set(size_bytes);
+                METRICS_BAL.size_bytes_histogram.observe(size_bytes);
+                METRICS_BAL.account_count.set(account_count as i64);
+                METRICS_BAL.slot_count.set(slot_count as i64);
+            }
+        );
 
         Ok((produced_bal, witness, result))
     }

crates/networking/p2p/sync/full.rs

@@ -790,7 +790,7 @@ async fn run_blocks_pipeline(
         for (block, bal) in blocks.into_iter().zip(bals.into_iter()) {
             let block_hash = block.hash();
             blockchain
-                .add_block_pipeline(block, bal.as_ref())
+                .add_block_pipeline(block, bal.map(Arc::new))
                 .map_err(|e| {
                     (
                         e,

crates/networking/rpc/rpc.rs

@@ -452,12 +452,12 @@ pub fn start_block_executor(blockchain: Arc<Blockchain>) -> UnboundedSender<Bloc
         .spawn(move || {
             while let Some((notify, block, bal, make_witness)) = block_receiver.blocking_recv() {
                 let result = (|| {
+                    let bal = bal.map(Arc::new);
                     if make_witness {
-                        let witness =
-                            blockchain.add_block_pipeline_with_witness(block, bal.as_ref())?;
+                        let witness = blockchain.add_block_pipeline_with_witness(block, bal)?;
                         Ok(Some(witness))
                     } else {
-                        blockchain.add_block_pipeline(block, bal.as_ref())?;
+                        blockchain.add_block_pipeline(block, bal)?;
                         Ok(None)
                     }
                 })();

crates/vm/backends/levm/mod.rs

@@ -71,7 +71,6 @@ use rayon::iter::{IntoParallelIterator, IntoParallelRefIterator, ParallelIterato
 #[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
 use rustc_hash::{FxHashMap, FxHashSet};
 use std::cmp::min;
-#[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
 use std::sync::Arc;
 #[cfg(all(feature = "rayon", not(feature = "eip-8025")))]
 use std::sync::atomic::AtomicBool;
@@ -397,7 +396,7 @@ impl LEVM {
         merkleizer: Option<Sender<Vec<AccountUpdate>>>,
         queue_length: &AtomicUsize,
         crypto: &dyn Crypto,
-        header_bal: Option<&BlockAccessList>,
+        header_bal: Option<Arc<BlockAccessList>>,
         bal_parallel_exec_enabled: bool,
     ) -> Result<(BlockExecutionResult, Option<BlockAccessList>), EvmError> {
         let chain_config = db.store.get_chain_config()?;
@@ -440,15 +439,15 @@ impl LEVM {
             // Note: size cap validation is deferred until after transaction processing
             // so that transaction-level errors (e.g. gas allowance exceeded) take
             // priority, matching the reference implementation's validation order.
-            validate_header_bal_indices(bal, block.body.transactions.len())
+            validate_header_bal_indices(&bal, block.body.transactions.len())
                 .map_err(|e| EvmError::Custom(e.to_string()))?;
 
             // Outer db has no BAL recorder: header BAL drives validation.
             // Per-tx tx_dbs enable a shadow recorder for accessed-entry checks.
             Self::prepare_block(block, db, vm_type, crypto)?;
 
             // Build validation index once — shared across parallel execution and post-exec seeding.
-            let validation_index = bal.build_validation_index();
+            let validation_index = Arc::new(bal.build_validation_index());
 
             // Drain system call state and snapshot for per-tx db seeding
             LEVM::get_state_transitions_tx(db)?;
@@ -459,12 +458,12 @@ impl LEVM {
                 &transactions_with_sender,
                 db,
                 vm_type,
-                bal,
+                Arc::clone(&bal),
                 merkleizer.as_ref(),
                 queue_length,
                 system_seed,
                 crypto,
-                &validation_index,
+                Arc::clone(&validation_index),
             );
 
             // If parallel execution failed (e.g. BAL validation), still check system
@@ -484,7 +483,7 @@ impl LEVM {
                         u32::try_from(block.body.transactions.len()).unwrap_or(u32::MAX);
                     if Self::seed_db_from_bal(
                         db,
-                        bal,
+                        &bal,
                         last_tx_idx,
                         &validation_index.accounts_by_min_index,
                     )
@@ -507,7 +506,7 @@ impl LEVM {
             // Eager seed retained: lazy_bal cursor is per-tx only; outer DB has no cursor.
             Self::seed_db_from_bal(
                 db,
-                bal,
+                &bal,
                 last_tx_idx,
                 &validation_index.accounts_by_min_index,
             )?;
@@ -533,7 +532,7 @@ impl LEVM {
             let withdrawal_idx = u32::try_from(block.body.transactions.len())
                 .map(|n| n.saturating_add(1))
                 .unwrap_or(u32::MAX);
-            Self::validate_bal_withdrawal_index(db, bal, withdrawal_idx, &validation_index)?;
+            Self::validate_bal_withdrawal_index(db, &bal, withdrawal_idx, &validation_index)?;
 
             // Mark storage_reads that occurred during the withdrawal/request phase.
             if !unread_storage_reads.is_empty() {
@@ -584,7 +583,7 @@ impl LEVM {
 
             // EIP-7928 size cap: validated after execution so that transaction-level
             // errors (e.g. gas allowance exceeded) take priority.
-            validate_block_access_list_size(&block.header, &chain_config, bal)
+            validate_block_access_list_size(&block.header, &chain_config, &bal)
                 .map_err(|e| EvmError::Custom(e.to_string()))?;
 
             return Ok((
@@ -1013,12 +1012,12 @@ impl LEVM {
         txs_with_sender: &[(&Transaction, Address)],
         db: &mut GeneralizedDatabase,
         vm_type: VMType,
-        bal: &BlockAccessList,
+        bal: Arc<BlockAccessList>,
         merkleizer: Option<&Sender<Vec<AccountUpdate>>>,
         queue_length: &AtomicUsize,
         system_seed: Arc<CacheDB>,
         crypto: &dyn Crypto,
-        validation_index: &BalAddressIndex,
+        validation_index: Arc<BalAddressIndex>,
     ) -> Result<
         (
             Vec<Receipt>,
@@ -1053,7 +1052,7 @@ impl LEVM {
         // Skipped when the caller merkleizes optimistically from the input BAL; the
         // conversion is then redundant work (and does pre-state reads we don't need).
         if let Some(merkleizer) = merkleizer {
-            let account_updates = Self::bal_to_account_updates(bal, store.as_ref())?;
+            let account_updates = Self::bal_to_account_updates(&bal, store.as_ref())?;
             merkleizer
                 .send(account_updates)
                 .map_err(|e| EvmError::Custom(format!("merkleizer send failed: {e}")))?;
@@ -1100,9 +1099,9 @@ impl LEVM {
                 .is_some_and(|a| a.storage.contains_key(key))
         });
 
-        // Small capacity hint — per-tx DBs materialize only touched accounts via lazy_bal cursor.
-        let arc_bal = Arc::new(bal.clone());
-        let arc_idx = Arc::new(validation_index.clone());
+        // Already-owned Arcs from the caller; per-thread/per-tx uses below are pointer clones.
+        let arc_bal = bal;
+        let arc_idx = validation_index;
 
         // 2. Execute all txs in parallel (embarrassingly parallel, BAL-seeded).
         //    BAL validation runs INSIDE the par_iter closure (parallel) but its
@@ -1131,6 +1130,7 @@ impl LEVM {
             .into_par_iter()
             .map(|tx_idx| -> Result<_, EvmError> {
                 let (tx, sender) = &txs_with_sender[tx_idx];
+                // Small capacity hint — per-tx DBs materialize only touched accounts via lazy_bal cursor.
                 let mut tx_db = GeneralizedDatabase::new_with_shared_base_and_capacity(
                     store.clone(),
                     system_seed.clone(),
@@ -1230,8 +1230,8 @@ impl LEVM {
                         seed_idx,
                         &current_state,
                         &codes,
-                        bal,
-                        validation_index,
+                        &arc_bal,
+                        &arc_idx,
                         &system_seed,
                         &store,
                     )
@@ -1241,19 +1241,19 @@ impl LEVM {
 
                     // EIP-7928 (Group B): missing-access detection via shadow recorder.
                     for addr in &shadow_touched {
-                        if !validation_index.addr_to_idx.contains_key(addr) {
+                        if !arc_idx.addr_to_idx.contains_key(addr) {
                             return Err(EvmError::Custom(format!(
                                 "BAL validation failed for tx {tx_idx}: account {addr:?} was \
                                  accessed during execution but is missing from BAL"
                             )));
                         }
                     }
                     for (addr, slot) in &shadow_reads {
-                        let Some(&bal_acct_idx) = validation_index.addr_to_idx.get(addr) else {
+                        let Some(&bal_acct_idx) = arc_idx.addr_to_idx.get(addr) else {
                             // Already caught by the touched-address check above.
                             continue;
                         };
-                        let acct = &bal.accounts()[bal_acct_idx];
+                        let acct = &arc_bal.accounts()[bal_acct_idx];
                         let in_changes = acct
                             .storage_changes
                             .binary_search_by(|sc| sc.slot.cmp(slot))

crates/vm/backends/mod.rs

@@ -111,7 +111,7 @@ impl Evm {
         block: &Block,
         merkleizer: Option<Sender<Vec<AccountUpdate>>>,
         queue_length: &AtomicUsize,
-        bal: Option<&BlockAccessList>,
+        bal: Option<Arc<BlockAccessList>>,
         bal_parallel_exec_enabled: bool,
     ) -> Result<(BlockExecutionResult, Option<BlockAccessList>), EvmError> {
         LEVM::execute_block_pipeline(