perf(l1): batch and stream the BAL code prefetch

CHANGELOG.md

@@ -15,6 +15,7 @@
 ### 2026-08-04
 
 - Cut the cost of a cold contract-code access: store jump destinations as a 1-bit-per-byte bitmap instead of a persisted RLP list of `u32` offsets, count the bytecode in the code cache's byte budget, answer `EXTCODESIZE` from the code-length table instead of materializing the bytecode, and give the account-code column families a bloom filter (4KB data blocks on the blob-backed bytecode CF). Raises the code cache's byte budget from an effective 64 MiB of jump tables to 256 MiB of bytecode, and bumps the store schema version so an older binary warns rather than failing on the new value format. `COLD_ACCOUNT_CODE_ACCESS` drops from 7736 to 4652 gas in the EIP-8038 repricing fit, and `COLD_ACCOUNT_CODE_WRITE` from 10415 to 6355 [#7095](https://github.com/lambdaclass/ethrex/pull/7095)
+- Batch and stream the BAL contract-code prefetch: warm accounts and their code in chunks instead of reading every access-list account before the first bytecode, take code hashes from the account read rather than a second lookup per account, and add a batched bytecode read that resolves the buffer and code cache first, then either fans out parallel point gets or shards the remainder across concurrent `multi_get`s, whichever reaches the greater read queue depth for the batch size on this host [#7099](https://github.com/lambdaclass/ethrex/pull/7099)
 
 ### 2026-07-22
 

crates/blockchain/vm.rs

@@ -130,6 +130,10 @@ impl StoreVmDatabase {
 }
 
 impl VmDatabase for StoreVmDatabase {
+    fn code_cache_budget_bytes(&self) -> u64 {
+        self.store.code_cache_budget_bytes()
+    }
+
     #[instrument(
         level = "trace",
         name = "Account read",
@@ -361,6 +365,44 @@ impl VmDatabase for StoreVmDatabase {
         }
     }
 
+    #[instrument(
+        level = "trace",
+        name = "Account codes batch read",
+        skip_all,
+        fields(namespace = "block_execution")
+    )]
+    fn get_account_codes_batch(&self, code_hashes: &[H256]) -> Result<Vec<Option<Code>>, EvmError> {
+        // The empty hash is answered here rather than sent to the store, matching
+        // `get_account_code`, so a batch containing EOAs does not fault on it.
+        let to_read: Vec<H256> = code_hashes
+            .iter()
+            .copied()
+            .filter(|h| *h != *EMPTY_KECCAK_HASH)
+            .collect();
+        let read = self
+            .store
+            .get_account_codes_batch(&to_read)
+            .map_err(|e| EvmError::DB(e.to_string()))?;
+
+        let mut by_hash: FxHashMap<H256, Code> = FxHashMap::default();
+        for (hash, code) in to_read.iter().zip(read.into_iter()) {
+            if let Some(code) = code {
+                by_hash.insert(*hash, code);
+            }
+        }
+
+        Ok(code_hashes
+            .iter()
+            .map(|h| {
+                if *h == *EMPTY_KECCAK_HASH {
+                    Some(Code::default())
+                } else {
+                    by_hash.get(h).cloned()
+                }
+            })
+            .collect())
+    }
+
     #[instrument(
         level = "trace",
         name = "Code metadata read",

crates/storage/store.rs

@@ -923,6 +923,19 @@ impl Store {
             .map_err(StoreError::from)
     }
 
+    /// Capacity of the bytecode cache, in [`Code::size`] bytes, which is how much
+    /// bytecode this node is configured to keep resident and so bounds what a block warm
+    /// reads speculatively.
+    ///
+    /// A poisoned cache reports zero, which stops warming rather than warming into a
+    /// cache no reader can reach.
+    pub fn code_cache_budget_bytes(&self) -> u64 {
+        self.account_code_cache
+            .lock()
+            .map(|cache| cache.max_size)
+            .unwrap_or(0)
+    }
+
     /// Get account code by its hash.
     ///
     /// Checks the in-memory block-data buffer first, then the LRU cache
@@ -966,6 +979,154 @@ impl Store {
         Ok(Some(code))
     }
 
+    /// Batched [`Self::get_account_code`].
+    ///
+    /// Resolves the buffer and the LRU first, then reads whatever is left by whichever
+    /// of two strategies gets more of those reads in flight for a batch this size: a
+    /// parallel fan-out of point gets, or sorted keys split into contiguous shards read
+    /// concurrently. See the comment on the read below for how the choice is made. The
+    /// LRU is locked once for the whole batch rather than twice per code.
+    ///
+    /// Results are returned in the order of `code_hashes`. Duplicate hashes are read
+    /// once. `None` means the hash is absent from the database.
+    pub fn get_account_codes_batch(
+        &self,
+        code_hashes: &[H256],
+    ) -> Result<Vec<Option<Code>>, StoreError> {
+        let mut out: Vec<Option<Code>> = vec![None; code_hashes.len()];
+        // Positions to fill per distinct hash, so a repeated hash costs one read.
+        let mut pending: HashMap<H256, Vec<usize>> = HashMap::new();
+
+        {
+            let buffer = self.buffer()?;
+            let mut cache = self
+                .account_code_cache
+                .lock()
+                .map_err(|_| StoreError::LockError)?;
+            for (i, hash) in code_hashes.iter().enumerate() {
+                if let Some(code) = buffer.get_code(hash) {
+                    out[i] = Some(code);
+                } else if let Some(code) = cache.get(hash)? {
+                    out[i] = Some(code);
+                } else {
+                    pending.entry(*hash).or_default().push(i);
+                }
+            }
+        }
+
+        if pending.is_empty() {
+            return Ok(out);
+        }
+
+        let mut missing: Vec<H256> = pending.keys().copied().collect();
+        missing.sort_unstable();
+
+        // Cold blob reads here are latency-bound, so what matters is how many are in
+        // flight. Two ways to get there, and which one wins depends on the batch size:
+        //
+        // * A parallel fan-out of point gets reaches queue depth ~= core count. Cheap
+        //   for any batch: rayon's pool is already warm.
+        // * Contiguous shards of the SORTED keys, one blocking thread each, reach queue
+        //   depth ~= shard count and share RocksDB blocks within a shard. This is the
+        //   only way past core count (async_io is OFF in this build, so a single
+        //   `multi_get` runs the whole batch at queue depth 1), but it pays a thread
+        //   spawn per shard.
+        //
+        // So shard only once sharding can actually beat the fan-out, i.e. once the batch
+        // is wide enough to form more shards than there are cores. Below that the
+        // fan-out is both deeper and cheaper, and a single serial `multi_get` would be
+        // far worse than either.
+        //
+        // The shard cap has to sit above the core count, or on a host with at least that
+        // many cores the fan-out would always win and this path would be unreachable.
+        // Twice the cores guarantees that, with a floor so a small host keeps the depth
+        // it can already reach: these threads block on I/O rather than compute, so more
+        // of them than cores is the point.
+        const KEYS_PER_SHARD: usize = 256;
+        const MIN_SHARD_CAP: usize = 64;
+        let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get());
+        let max_shards = parallelism.saturating_mul(2).max(MIN_SHARD_CAP);
+        let shards = missing.len().div_ceil(KEYS_PER_SHARD).min(max_shards);
+        let read_view = self.backend.begin_read()?;
+        // Both paths decode in whatever thread did the read, so rebuilding the jumpdest
+        // bitmap for a legacy entry stays off the caller's thread and stays parallel.
+        let decode = |hash: &H256, value: Option<Vec<u8>>| -> Result<Option<Code>, StoreError> {
+            let Some(bytes) = value else { return Ok(None) };
+            let (bytecode_slice, jumpdests) = decode_bytes(&bytes)?;
+            Ok(Some(Code::from_parts_unchecked(
+                *hash,
+                bytecode_slice,
+                decode_jumpdests(bytecode_slice, jumpdests)?,
+            )))
+        };
+        let decoded: Vec<Result<Option<Code>, StoreError>> = if shards > parallelism {
+            let chunk = missing.len().div_ceil(shards);
+            let rv = read_view.as_ref();
+            let read_shard = |hashes: &[H256]| -> Vec<Result<Option<Code>, StoreError>> {
+                let keys: Vec<&[u8]> = hashes.iter().map(|h| h.as_bytes()).collect();
+                rv.multi_get(ACCOUNT_CODES, &keys)
+                    .into_iter()
+                    .zip(hashes.iter())
+                    .map(|(value, hash)| decode(hash, value?))
+                    .collect()
+            };
+            std::thread::scope(|scope| {
+                let handles: Vec<_> = missing
+                    .chunks(chunk)
+                    .map(|ck| (ck, scope.spawn(move || read_shard(ck))))
+                    .collect();
+                handles
+                    .into_iter()
+                    .flat_map(|(shard, handle)| {
+                        // A panicked shard becomes an `Err` per key it covered, keeping
+                        // the results aligned with `missing` and leaving the caller's
+                        // best-effort handling to decide. Re-panicking here would
+                        // escalate a warm into taking down whatever runs it.
+                        handle.join().unwrap_or_else(|_| {
+                            shard
+                                .iter()
+                                .map(|_| {
+                                    Err(StoreError::Custom(
+                                        "account code shard panicked".to_string(),
+                                    ))
+                                })
+                                .collect()
+                        })
+                    })
+                    .collect()
+            })
+        } else {
+            // One key per read, so `get` rather than a single-key `multi_get`: the
+            // batched call sets up its own result buffers, which is pure overhead here.
+            let rv = read_view.as_ref();
+            missing
+                .par_iter()
+                .map(|hash| decode(hash, rv.get(ACCOUNT_CODES, hash.as_bytes())?))
+                .collect()
+        };
+
+        let mut fetched: Vec<Code> = Vec::new();
+        for (hash, code) in missing.iter().zip(decoded.into_iter()) {
+            let Some(code) = code? else { continue };
+            for &i in pending.get(hash).into_iter().flatten() {
+                out[i] = Some(code.clone());
+            }
+            fetched.push(code);
+        }
+
+        if !fetched.is_empty() {
+            let mut cache = self
+                .account_code_cache
+                .lock()
+                .map_err(|_| StoreError::LockError)?;
+            for code in &fetched {
+                cache.insert(code)?;
+            }
+        }
+
+        Ok(out)
+    }
+
     /// Check if account code exists by its hash, without constructing the full `Code` struct.
     /// More efficient than `get_account_code` for existence checks since it skips
     /// RLP decoding and `Code` struct construction (no jumpdest-bitmap decoding).

crates/vm/backends/levm/db.rs

@@ -79,6 +79,28 @@ impl LevmDatabase for DatabaseLogger {
         self.store.as_ref().get_account_code(code_hash)
     }
 
+    fn code_cache_budget_bytes(&self) -> u64 {
+        self.store.as_ref().code_cache_budget_bytes()
+    }
+
+    fn prefetch_codes(&self, code_hashes: &[CoreH256]) -> Result<u64, DatabaseError> {
+        // Record before delegating, exactly as the per-hash path does, so a batched
+        // read cannot leave the witness short of a bytecode it observed.
+        {
+            let mut code_accessed = self
+                .code_accessed
+                .lock()
+                .map_err(|_| DatabaseError::Custom("Could not lock mutex".to_string()))?;
+            code_accessed.extend(
+                code_hashes
+                    .iter()
+                    .filter(|h| **h != *EMPTY_KECCAK_HASH)
+                    .copied(),
+            );
+        }
+        self.store.as_ref().prefetch_codes(code_hashes)
+    }
+
     fn get_code_metadata(&self, code_hash: CoreH256) -> Result<CodeMetadata, DatabaseError> {
         // A size-only read still observes the bytecode, so the witness must carry it:
         // EIP-8025 stateless validation recomputes the length from the code itself, and
@@ -114,6 +136,10 @@ impl LevmDatabase for DynVmDatabase {
             .collect())
     }
 
+    fn code_cache_budget_bytes(&self) -> u64 {
+        <dyn VmDatabase>::code_cache_budget_bytes(self.as_ref())
+    }
+
     fn get_storage_value(
         &self,
         address: CoreAddress,
@@ -153,6 +179,14 @@ impl LevmDatabase for DynVmDatabase {
             .map_err(|e| DatabaseError::Custom(e.to_string()))
     }
 
+    fn get_account_codes_batch(
+        &self,
+        code_hashes: &[CoreH256],
+    ) -> Result<Vec<Option<Code>>, DatabaseError> {
+        <dyn VmDatabase>::get_account_codes_batch(self.as_ref(), code_hashes)
+            .map_err(|e| DatabaseError::Custom(e.to_string()))
+    }
+
     fn get_code_metadata(&self, code_hash: CoreH256) -> Result<CodeMetadata, DatabaseError> {
         <dyn VmDatabase>::get_code_metadata(self.as_ref(), code_hash)
             .map_err(|e| DatabaseError::Custom(e.to_string()))