perf(l1): move per-tx BAL validation into the par_iter closure (#6677)

## Summary

`execute_block_parallel` previously returned `(current_state, codes,
shadow_touched, shadow_reads)` per tx and validated them in a serial
post-`par_iter` loop across all txs. Validation is per-tx pure work;
only the marking of `unread_storage_reads` / `unaccessed_pure_accounts`
mutates cross-tx state.

This PR moves `validate_tx_execution` + shadow-touched / shadow-reads
checks into the rayon closure (parallel). The closure also precomputes
the small `(Vec<(Address, H256)>, Vec<Address>)` inputs the serial pass
uses to update the shared sets, so `current_state` + `codes` no longer
cross the rayon boundary.

BAL validation errors are deferred via `Option<EvmError>` in the result
tuple so the post-`par_iter` gas-limit check still takes priority
(preserves `GAS_USED_OVERFLOW` > BAL mismatch on blocks exceeding the
gas limit; the BAL is built assuming rejected txs, so miner balance in
the BAL won't match execution that ran all txs).

## Changes

- `TxExecResult` tuple shrinks from 8 fields (with two heavy
`FxHashMap`s) to 7 fields with `Vec`s only.
- Inside the closure, after `execute_tx_in_block`:
- Compute `reads_satisfied: Vec<(Address, H256)>` (touched storage keys
per non-destroyed account) and `destroyed: Vec<Address>` (selfdestructed
accounts whose remaining BAL storage_reads are wiped) from
`current_state`.
- Run `validate_tx_execution` + shadow-touched + shadow-reads checks;
capture the first error as `Option<EvmError>` (deferred).
  - Drop `current_state` and `codes` before returning.
- After `par_iter`:
- Existing 2D gas allowance / breakdowns / block-overflow loop
unchanged.
- **New step**: surface the first deferred BAL error in tx order (only
reached if the gas-limit check passes).
- Apply `reads_satisfied` / `destroyed` to `unread_storage_reads` and
`tracked_accounts` to `unaccessed_pure_accounts` (cheap hash-set ops).

## Invariants

1. **Gas-limit error priority preserved**: the serial pass runs gas / 2D
allowance / block-overflow before surfacing any deferred BAL error.
2. **Tx-order error reporting preserved**: errors are surfaced in sorted
`tx_idx` order; the first failing tx still wins.
3. **Destroyed-account semantics preserved**: the `Destroyed |
DestroyedModified` branch in the unread-reads marking is now an explicit
`Vec<Address>` from the closure; the serial pass does the same
`retain(|&(a, _)| a != addr)` it did before.
4. **No semantic change to validation**: same checks, same error
strings, only the location moves.

## Benchmark

Fixture: `bal-devnet-7-mainnet-mix-460` (460 blocks, ~30 Ggas,
transfer/EVM mix). `release-with-debug`, `import-bench --with-bal`.
Baseline = `feat/import-bench-bal-tooling` (bench tooling only, no perf
change).

| metric           | baseline | this PR | delta              |
|------------------|---------:|--------:|--------------------|
| wall time        | 8.575 s  | 7.871 s | **-704 ms (-8.2%)** |
| agg Ggas/s       | 3.901    | 4.291   | **+10.0%**         |
| avg ms / block   | 16.884   | 15.349  | **-1.54 ms (-9.1%)** |
| p95 ms / block   | 17.730   | 16.050  | -1.68 ms (-9.5%)   |
| max ms / block   | 90.060   | 85.000  | -5.06 ms           |
| **exec avg** | **15.574 ms** | **14.172 ms** | **-1.40 ms (-9.0%)** |
| merkle           | 0.479 ms | 0.412 ms | flat (run variance) |
| store            | 0.667 ms | 0.610 ms | flat (run variance) |
| warmer           | 1.365 ms | 1.315 ms | flat               |

The win is concentrated in `exec` (-9%), which is exactly where the
serial post-`par_iter` validation pass used to live. The merkle / store
/ warmer deltas are within run-to-run variance; this PR doesn't touch
those phases.

Compare dashboard:
<https://edgl.dev/share/compare-feat-import-bench-bal-tooling@02a5663c-vs-perf-bal-validation-into-par-iter@cc8ba27c.html>

## Test plan

- [x] `cargo check --bin ethrex` (default features)
- [x] `cargo fmt --all --check`
- [x] `cargo clippy --workspace --no-deps -- -D warnings`
- [ ] EF blockchain tests (Amsterdam; two-pass parallel runner)
- [ ] Hive `bal` group

Author: edg-l

CHANGELOG.md

@@ -23,6 +23,7 @@
 ### 2026-05-19
 
 - Lazy BAL cursor for per-tx parallel execution [#6669](https://github.com/lambdaclass/ethrex/pull/6669)
+- Move per-tx BAL validation into the rayon par_iter closure on the parallel execution path [#6677](https://github.com/lambdaclass/ethrex/pull/6677)
 
 ### 2026-05-15
 

crates/vm/backends/levm/mod.rs

@@ -1097,17 +1097,26 @@ impl LEVM {
         let arc_idx = Arc::new(validation_index.clone());
 
         // 2. Execute all txs in parallel (embarrassingly parallel, BAL-seeded).
-        //    BAL validation is deferred to after the gas limit check (step 3) so that
-        //    blocks exceeding gas limit produce GAS_USED_OVERFLOW before BAL mismatch.
+        //    BAL validation runs INSIDE the par_iter closure (parallel) but its
+        //    errors are deferred via Option<EvmError> so the post-par_iter
+        //    gas-limit check still takes priority (GAS_USED_OVERFLOW must beat
+        //    BAL mismatch on blocks exceeding the gas limit; the BAL is built
+        //    assuming rejected txs, so miner balance in the BAL won't match
+        //    execution that ran all txs).
+        //
+        //    The closure also precomputes the small (Vec<(Address, H256)>,
+        //    Vec<Address>) inputs needed to update the shared
+        //    `unread_storage_reads` / `unaccessed_pure_accounts` sets, so the
+        //    serial pass after par_iter is just hash-set ops; current_state
+        //    and codes never cross the rayon boundary.
         type TxExecResult = (
             usize,
             TxType,
             ExecutionReport,
-            FxHashMap<Address, LevmAccount>,
-            FxHashMap<H256, ethrex_common::types::Code>,
             FxHashSet<Address>,   // accessed_accounts tracker (coarse)
-            Vec<Address>,         // shadow recorder touched_addresses (EIP-7928 exact)
-            Vec<(Address, U256)>, // shadow recorder storage_reads (EIP-7928 exact)
+            Vec<(Address, H256)>, // reads_satisfied: (addr, slot) loaded during this tx
+            Vec<Address>,         // destroyed: accounts selfdestructed during this tx
+            Option<EvmError>,     // deferred BAL validation error
         );
 
         let exec_results: Result<Vec<TxExecResult>, EvmError> = (0..n_txs)
@@ -1173,37 +1182,120 @@ impl LEVM {
                     .take()
                     .map(|mut r| (r.take_touched_addresses(), r.take_storage_reads()))
                     .unwrap_or_default();
+
+                // Precompute the per-tx inputs the serial pass uses to update
+                // the shared unread_storage_reads set. Selfdestruct clears
+                // storage from the final state, so destroyed accounts
+                // satisfy ALL their BAL storage_reads regardless of which
+                // slots remain in `current_state`.
+                // Rough avg storage slots per touched account; over-allocation
+                // is cheap compared to 2-3 reallocations on the hot path.
+                let mut reads_satisfied: Vec<(Address, H256)> =
+                    Vec::with_capacity(current_state.len() * 4);
+                // `destroyed` stays empty on the typical block (selfdestruct
+                // is rare post-EIP-6780), so `Vec::new()` (no allocation) is
+                // optimal here.
+                let mut destroyed: Vec<Address> = Vec::new();
+                for (addr, acct) in &current_state {
+                    if matches!(
+                        acct.status,
+                        AccountStatus::Destroyed | AccountStatus::DestroyedModified
+                    ) {
+                        destroyed.push(*addr);
+                    } else {
+                        for key in acct.storage.keys() {
+                            reads_satisfied.push((*addr, *key));
+                        }
+                    }
+                }
+
+                // Run BAL validation inline. Errors are DEFERRED: stored in
+                // Option<EvmError> so the serial gas-limit check below still
+                // takes priority. Borrow current_state / codes during the
+                // validation closure, then drop them before returning so
+                // they don't cross the rayon boundary.
+                let deferred_bal_err: Option<EvmError> = (|| -> Result<(), EvmError> {
+                    let bal_idx = u32::try_from(tx_idx + 1).unwrap_or(u32::MAX);
+                    let seed_idx = u32::try_from(tx_idx).unwrap_or(u32::MAX);
+                    Self::validate_tx_execution(
+                        bal_idx,
+                        seed_idx,
+                        &current_state,
+                        &codes,
+                        bal,
+                        validation_index,
+                        &system_seed,
+                        &store,
+                    )
+                    .map_err(|e| {
+                        EvmError::Custom(format!("BAL validation failed for tx {tx_idx}: {e}"))
+                    })?;
+
+                    // EIP-7928 (Group B): missing-access detection via shadow recorder.
+                    for addr in &shadow_touched {
+                        if !validation_index.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 {
+                            // Already caught by the touched-address check above.
+                            continue;
+                        };
+                        let acct = &bal.accounts()[bal_acct_idx];
+                        let in_changes = acct
+                            .storage_changes
+                            .binary_search_by(|sc| sc.slot.cmp(slot))
+                            .is_ok();
+                        let in_reads = acct.storage_reads.contains(slot);
+                        if !in_changes && !in_reads {
+                            return Err(EvmError::Custom(format!(
+                                "BAL validation failed for tx {tx_idx}: storage slot {slot} of \
+                                 account {addr:?} was read during execution but is missing from \
+                                 BAL (no storage_changes or storage_reads entry)"
+                            )));
+                        }
+                    }
+                    Ok(())
+                })()
+                .err();
+
+                drop(current_state);
+                drop(codes);
+
                 Ok((
                     tx_idx,
                     tx.tx_type(),
                     report,
-                    current_state,
-                    codes,
                     tracked,
-                    shadow_touched,
-                    shadow_reads,
+                    reads_satisfied,
+                    destroyed,
+                    deferred_bal_err,
                 ))
             })
             .collect();
 
         let mut exec_results = exec_results?;
 
-        // Sort so gas accounting and validation happen in tx order.
-        exec_results.sort_unstable_by_key(|(idx, _, _, _, _, _, _, _)| *idx);
-
-        // 3. Gas limit check — must happen BEFORE BAL validation so that blocks
-        //    exceeding the gas limit produce GAS_USED_OVERFLOW instead of a BAL
-        //    mismatch error (the BAL is built assuming rejected txs, so the miner
-        //    balance in the BAL won't match execution that ran all txs).
-        //
-        //    EIP-8037 PR #2703: also enforce the per-tx 2D inclusion check
-        //    against running block totals. A tx whose worst-case regular or
-        //    state contribution exceeds the remaining budget at its inclusion
-        //    position invalidates the block with GAS_ALLOWANCE_EXCEEDED.
+        // `IndexedParallelIterator` (via `(0..n_txs).into_par_iter()`) preserves
+        // source-index order through `.map().collect()`, so `exec_results` is
+        // already sorted. The sort is kept as a defensive guard against a future
+        // refactor swapping in an unordered iterator; `sort_unstable_by_key` on
+        // an already-sorted slice is near-linear via pdqsort, so the cost is
+        // negligible.
+        exec_results.sort_unstable_by_key(|(idx, _, _, _, _, _, _)| *idx);
+
+        // 3. Gas limit check — must happen BEFORE BAL validation errors so that
+        //    blocks exceeding the gas limit produce GAS_USED_OVERFLOW instead of
+        //    a BAL mismatch error. EIP-8037 PR #2703: also enforce the per-tx
+        //    2D inclusion check against running block totals.
         let mut block_regular_gas_used = 0_u64;
         let mut block_state_gas_used = 0_u64;
         let mut tx_gas_breakdowns: Vec<TxGasBreakdown> = Vec::with_capacity(exec_results.len());
-        for (tx_idx, _, report, _, _, _, _, _) in &exec_results {
+        for (tx_idx, _, report, _, _, _, _) in &exec_results {
             let (tx, _) = txs_with_sender
                 .get(*tx_idx)
                 .ok_or_else(|| EvmError::Custom(format!("tx index {tx_idx} out of bounds")))?;
@@ -1234,48 +1326,25 @@ impl LEVM {
             )));
         }
 
-        // 4. Per-tx BAL validation — now safe to run after gas limit is confirmed OK.
-        //    Also mark off storage_reads that appear in per-tx execution state.
-        for (tx_idx, _, _, current_state, codes, tracked_accounts, shadow_touched, shadow_reads) in
-            &exec_results
-        {
-            let bal_idx = u32::try_from(*tx_idx + 1).unwrap_or(u32::MAX);
-            let seed_idx = u32::try_from(*tx_idx).unwrap_or(u32::MAX);
-            Self::validate_tx_execution(
-                bal_idx,
-                seed_idx,
-                current_state,
-                codes,
-                bal,
-                validation_index,
-                &system_seed,
-                &store,
-            )
-            .map_err(|e| EvmError::Custom(format!("BAL validation failed for tx {tx_idx}: {e}")))?;
-
-            // Mark storage_reads that were actually loaded during this tx.
-            // storage_reads slots are NOT in storage_changes (conflict check ensures this),
-            // so they're not seeded. If a slot appears in the per-tx state's storage,
-            // the tx genuinely read it via SLOAD.
-            // Special case: selfdestruct clears storage from the final state, so reads
-            // that happened before destruction are no longer visible. For destroyed
-            // accounts, mark ALL their BAL storage_reads as satisfied.
+        // 4. Surface the first deferred BAL validation error (in tx order) now
+        //    that the gas-limit check has passed.
+        for (_, _, _, _, _, _, deferred) in &mut exec_results {
+            if let Some(err) = deferred.take() {
+                return Err(err);
+            }
+        }
+
+        // 5. Apply per-tx reads_satisfied / destroyed / tracked to the shared
+        //    sets (cheap hash-set ops; preserves prior semantics).
+        for (_, _, _, tracked_accounts, reads_satisfied, destroyed, _) in &exec_results {
             if !unread_storage_reads.is_empty() {
-                for (addr, acct) in current_state {
-                    if matches!(
-                        acct.status,
-                        AccountStatus::Destroyed | AccountStatus::DestroyedModified
-                    ) {
-                        unread_storage_reads.retain(|&(a, _)| a != *addr);
-                    } else {
-                        for key in acct.storage.keys() {
-                            unread_storage_reads.remove(&(*addr, *key));
-                        }
-                    }
+                for addr in destroyed {
+                    unread_storage_reads.retain(|&(a, _)| a != *addr);
+                }
+                for pair in reads_satisfied {
+                    unread_storage_reads.remove(pair);
                 }
             }
-
-            // Mark pure-access accounts that were accessed during this tx.
             // The coinbase is always accessed during fee finalization (geth's
             // readerTracker records it), even when the miner fee is zero and
             // ethrex skips the load_account call.
@@ -1285,44 +1354,12 @@ impl LEVM {
                     unaccessed_pure_accounts.remove(addr);
                 }
             }
-
-            // EIP-7928 (Group B): missing-access detection using the shadow recorder.
-            // For each address the per-tx shadow recorder marked as touched, the header
-            // BAL must contain an entry for it. For each storage read, the header BAL
-            // must carry the slot either in storage_changes or storage_reads.
-            for addr in shadow_touched {
-                if !validation_index.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 {
-                    // Already caught by the touched-address check above.
-                    continue;
-                };
-                let acct = &bal.accounts()[bal_acct_idx];
-                let in_changes = acct
-                    .storage_changes
-                    .binary_search_by(|sc| sc.slot.cmp(slot))
-                    .is_ok();
-                let in_reads = acct.storage_reads.contains(slot);
-                if !in_changes && !in_reads {
-                    return Err(EvmError::Custom(format!(
-                        "BAL validation failed for tx {tx_idx}: storage slot {slot} of \
-                         account {addr:?} was read during execution but is missing from \
-                         BAL (no storage_changes or storage_reads entry)"
-                    )));
-                }
-            }
         }
 
-        // 5. Build receipts in tx order.
+        // 6. Build receipts in tx order.
         let mut receipts = Vec::with_capacity(n_txs);
         let mut cumulative_gas_used = 0_u64;
-        for (_, tx_type, report, _, _, _, _, _) in exec_results {
+        for (_, tx_type, report, _, _, _, _) in exec_results {
             cumulative_gas_used += report.gas_spent;
             let receipt = Receipt::new(
                 tx_type,