chore(mdbx): replace deprecated MDBX_NOTLS with MDBX_NOSTICKYTHREADS (#23378 )

feat: customizable discovery defaults (#23843 )
Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>
2026-04-30 03:01:58 -04:00 · 2026-04-30 03:19:09 +00:00 · 2026-04-29 22:59:10 +00:00 · 2026-04-29 15:29:42 +00:00 · 2026-04-29 14:36:44 +00:00 · 2026-04-29 12:57:05 +00:00
56 changed files with 1473 additions and 859 deletions
--- a/.github/scripts/hive/build_simulators.sh
+++ b/.github/scripts/hive/build_simulators.sh
@@ -5,8 +5,8 @@ fixture_variant="${1:-osaka}"

 case "${fixture_variant}" in
    amsterdam)
-        eels_fixtures="https://github.com/ethereum/execution-spec-tests/releases/download/bal@v6.0.0/fixtures_bal.tar.gz"
-        eels_branch="devnets/snøbal/4"
+        eels_fixtures="https://github.com/ethereum/execution-spec-tests/releases/download/snobal-devnet-5@v8037.0.0/fixtures_snobal-devnet-5.tar.gz"
+        eels_branch="devnets/snobal/5"
        ;;
    osaka)
        eels_fixtures="https://github.com/ethereum/execution-spec-tests/releases/download/v5.3.0/fixtures_develop.tar.gz"
--- a/.github/workflows/hive.yml
+++ b/.github/workflows/hive.yml
@@ -6,9 +6,6 @@ on:
  workflow_dispatch:
  schedule:
    - cron: "0 0 * * *"
-  pull_request:
-    branches:
-      - "**"

 env:
  CARGO_TERM_COLOR: always
@@ -31,9 +28,9 @@ jobs:
    secrets: inherit

  prepare-hive:
-    if: github.repository == 'paradigmxyz/reth-oss' || github.repository == 'paradigmxyz/reth'
+    if: github.repository == 'paradigmxyz/reth'
    timeout-minutes: 45
-    runs-on: ${{ (github.repository == 'paradigmxyz/reth-oss' || github.repository == 'paradigmxyz/reth') && 'depot-ubuntu-latest-16' || 'ubuntu-latest' }}
+    runs-on: ${{ github.repository == 'paradigmxyz/reth' && 'depot-ubuntu-latest-16' || 'ubuntu-latest' }}
    permissions:
      contents: read
    strategy:
@@ -204,7 +201,7 @@ jobs:
      - prepare-hive
    name: Hive-Amsterdam / ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
    # Use larger runners for eels tests to avoid OOM runner crashes
-    runs-on: ${{ (github.repository == 'paradigmxyz/reth-oss' || github.repository == 'paradigmxyz/reth') && (contains(matrix.scenario.sim, 'eels') && 'depot-ubuntu-latest-8' || 'depot-ubuntu-latest-4') || 'ubuntu-latest' }}
+    runs-on: ${{ github.repository == 'paradigmxyz/reth' && (contains(matrix.scenario.sim, 'eels') && 'depot-ubuntu-latest-8' || 'depot-ubuntu-latest-4') || 'ubuntu-latest' }}
    permissions:
      contents: read
      issues: write
@@ -384,7 +381,7 @@ jobs:
      - prepare-hive
    name: Hive-Osaka / ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
    # Use larger runners for eels tests to avoid OOM runner crashes
-    runs-on: ${{ (github.repository == 'paradigmxyz/reth-oss' || github.repository == 'paradigmxyz/reth') && (contains(matrix.scenario.sim, 'eels') && 'depot-ubuntu-latest-8' || 'depot-ubuntu-latest-4') || 'ubuntu-latest' }}
+    runs-on: ${{ github.repository == 'paradigmxyz/reth' && (contains(matrix.scenario.sim, 'eels') && 'depot-ubuntu-latest-8' || 'depot-ubuntu-latest-4') || 'ubuntu-latest' }}
    permissions:
      contents: read
      issues: write
--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,5 +1,5 @@
 [workspace.package]
-version = "2.1.0"
+version = "2.2.0"
 edition = "2024"
 rust-version = "1.93"
 license = "MIT OR Apache-2.0"
@@ -433,14 +433,14 @@ reth-trie-sparse = { path = "crates/trie/sparse", default-features = false }
 reth-zstd-compressors = { version = "0.3.1", default-features = false }

 # revm
-revm = { version = "=38.0.0", default-features = false }
-revm-bytecode = { version = "=10.0.0", default-features = false }
-revm-database = { version = "=13.0.1", default-features = false }
-revm-state = { version = "=11.0.1", default-features = false }
-revm-primitives = { version = "=23.0.0", default-features = false }
-revm-interpreter = { version = "=35.0.1", default-features = false }
-revm-database-interface = { version = "=11.0.1", default-features = false }
-revm-inspectors = "=0.39.0"
+revm = { version = "38.0.0", default-features = false }
+revm-bytecode = { version = "10.0.0", default-features = false }
+revm-database = { version = "13.0.0", default-features = false }
+revm-state = { version = "11.0.0", default-features = false }
+revm-primitives = { version = "23.0.0", default-features = false }
+revm-interpreter = { version = "35.0.0", default-features = false }
+revm-database-interface = { version = "11.0.0", default-features = false }
+revm-inspectors = "0.39.0"

 # eth
 alloy-dyn-abi = "1.5.6"
@@ -456,33 +456,33 @@ alloy-trie = { version = "0.9.4", default-features = false }

 alloy-hardforks = "0.4.7"

-alloy-consensus = { version = "2.0.1", default-features = false }
-alloy-contract = { version = "2.0.1", default-features = false }
-alloy-eips = { version = "2.0.1", default-features = false }
-alloy-genesis = { version = "2.0.1", default-features = false }
-alloy-json-rpc = { version = "2.0.1", default-features = false }
-alloy-network = { version = "2.0.1", default-features = false }
-alloy-network-primitives = { version = "2.0.1", default-features = false }
-alloy-provider = { version = "2.0.1", features = ["reqwest", "debug-api"], default-features = false }
-alloy-pubsub = { version = "2.0.1", default-features = false }
-alloy-rpc-client = { version = "2.0.1", default-features = false }
-alloy-rpc-types = { version = "2.0.1", features = ["eth"], default-features = false }
-alloy-rpc-types-admin = { version = "2.0.1", default-features = false }
-alloy-rpc-types-anvil = { version = "2.0.1", default-features = false }
-alloy-rpc-types-beacon = { version = "2.0.1", default-features = false }
-alloy-rpc-types-debug = { version = "2.0.1", default-features = false }
-alloy-rpc-types-engine = { version = "2.0.1", default-features = false }
-alloy-rpc-types-eth = { version = "2.0.1", default-features = false }
-alloy-rpc-types-mev = { version = "2.0.1", default-features = false }
-alloy-rpc-types-trace = { version = "2.0.1", default-features = false }
-alloy-rpc-types-txpool = { version = "2.0.1", default-features = false }
-alloy-serde = { version = "2.0.1", default-features = false }
-alloy-signer = { version = "2.0.1", default-features = false }
-alloy-signer-local = { version = "2.0.1", default-features = false }
-alloy-transport = { version = "2.0.1" }
-alloy-transport-http = { version = "2.0.1", features = ["reqwest-rustls-tls"], default-features = false }
-alloy-transport-ipc = { version = "2.0.1", default-features = false }
-alloy-transport-ws = { version = "2.0.1", default-features = false }
+alloy-consensus = { version = "2.0.4", default-features = false }
+alloy-contract = { version = "2.0.4", default-features = false }
+alloy-eips = { version = "2.0.4", default-features = false }
+alloy-genesis = { version = "2.0.4", default-features = false }
+alloy-json-rpc = { version = "2.0.4", default-features = false }
+alloy-network = { version = "2.0.4", default-features = false }
+alloy-network-primitives = { version = "2.0.4", default-features = false }
+alloy-provider = { version = "2.0.4", features = ["reqwest", "debug-api"], default-features = false }
+alloy-pubsub = { version = "2.0.4", default-features = false }
+alloy-rpc-client = { version = "2.0.4", default-features = false }
+alloy-rpc-types = { version = "2.0.4", features = ["eth"], default-features = false }
+alloy-rpc-types-admin = { version = "2.0.4", default-features = false }
+alloy-rpc-types-anvil = { version = "2.0.4", default-features = false }
+alloy-rpc-types-beacon = { version = "2.0.4", default-features = false }
+alloy-rpc-types-debug = { version = "2.0.4", default-features = false }
+alloy-rpc-types-engine = { version = "2.0.4", default-features = false }
+alloy-rpc-types-eth = { version = "2.0.4", default-features = false }
+alloy-rpc-types-mev = { version = "2.0.4", default-features = false }
+alloy-rpc-types-trace = { version = "2.0.4", default-features = false }
+alloy-rpc-types-txpool = { version = "2.0.4", default-features = false }
+alloy-serde = { version = "2.0.4", default-features = false }
+alloy-signer = { version = "2.0.4", default-features = false }
+alloy-signer-local = { version = "2.0.4", default-features = false }
+alloy-transport = { version = "2.0.4" }
+alloy-transport-http = { version = "2.0.4", features = ["reqwest-rustls-tls"], default-features = false }
+alloy-transport-ipc = { version = "2.0.4", default-features = false }
+alloy-transport-ws = { version = "2.0.4", default-features = false }

 # misc
 either = { version = "1.15.0", default-features = false }
@@ -700,24 +700,3 @@ vergen-git2 = "9.1.0"

 # networking
 ipnet = "2.11"
-
-[patch.crates-io]
-revm = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-bytecode = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-context = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-context-interface = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-database = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-database-interface = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-handler = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-inspector = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-interpreter = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-precompile = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-primitives = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-state = { git = "https://github.com/bluealloy/revm", rev = "3ed3bdfed9ad6e5ba37f4e1f015436ab89ca98be" }
-revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "5eebb56819ee6bec5bfbc69a415276ee1a784fec" }
-alloy-evm = { git = "https://github.com/alloy-rs/evm", branch = "bal-devnet-4" }
-reth-codecs = { git = "https://github.com/paradigmxyz/reth-core", rev = "8612239c4f3dda83cc389f577b9eb04f10ebf81d" }
-reth-codecs-derive = { git = "https://github.com/paradigmxyz/reth-core", rev = "8612239c4f3dda83cc389f577b9eb04f10ebf81d" }
-reth-primitives-traits = { git = "https://github.com/paradigmxyz/reth-core", rev = "8612239c4f3dda83cc389f577b9eb04f10ebf81d" }
-reth-rpc-traits = { git = "https://github.com/paradigmxyz/reth-core", rev = "8612239c4f3dda83cc389f577b9eb04f10ebf81d" }
-reth-zstd-compressors = { git = "https://github.com/paradigmxyz/reth-core", rev = "8612239c4f3dda83cc389f577b9eb04f10ebf81d" }
--- a/bin/reth-bench/src/bench/generate_big_block.rs
+++ b/bin/reth-bench/src/bench/generate_big_block.rs
@@ -21,6 +21,7 @@ use alloy_rpc_types_engine::{
 };
 use clap::Parser;
 use eyre::Context;
+use futures::{stream, StreamExt};
 use reth_chainspec::EthChainSpec;
 use reth_cli::chainspec::ChainSpecParser;
 use reth_cli_runner::CliContext;
@@ -270,6 +271,15 @@ pub struct Command {
    /// the flattened BAL on the stored payload.
    #[arg(long, default_value_t = false)]
    bal: bool,
+
+    /// Maximum number of in-flight RPC fetches to keep buffered ahead of the merger.
+    ///
+    /// Each entry is one full per-block fetch (block + receipts, plus BAL when `--bal` is
+    /// set). Larger values absorb RPC latency at the cost of more concurrent connections
+    /// and memory; the buffer persists across `--num-big-blocks` so prefetching continues
+    /// across big-block boundaries.
+    #[arg(long, value_name = "PREFETCH_BUFFER", default_value_t = 32)]
+    prefetch_buffer: usize,
 }

 impl Command {
@@ -322,13 +332,27 @@ impl Command {
        }
        let mut prev_big_block_header: Option<PrevBigBlockHeader> = None;

-        // Track the next block to fetch across big blocks so they don't overlap.
+        // Persistent prefetch stream: keeps `prefetch_buffer` per-block fetches in flight
+        // ahead of the merger across all big blocks. Each item is a fully materialized
+        // `FetchedBlock` (or `None` once the chain tip is reached on this fetch).
+        let prefetch_buffer = self.prefetch_buffer.max(1);
+        let bal_enabled = self.bal;
+        let block_stream = stream::iter(self.from_block..)
+            .map(|block_number| {
+                let provider = provider.clone();
+                async move { fetch_one_block(provider, block_number, bal_enabled).await }
+            })
+            .buffered(prefetch_buffer);
+        let mut block_stream = Box::pin(block_stream);
+
+        // Track the next block number we expect from the stream (purely for logging /
+        // big-block range bookkeeping; the stream produces blocks in `from_block..` order).
        let mut next_block = self.from_block;

        for big_block_idx in 0..self.num_big_blocks {
            let range_start = next_block;

-            // Fetch consecutive blocks until the gas target is reached.
+            // Drain the prefetch stream until the gas target is reached for this big block.
            let mut blocks = Vec::new();
            let mut block_receipts: Vec<Vec<Receipt>> = Vec::new();
            let mut block_access_lists: Vec<Option<BlockAccessList>> = Vec::new();
@@ -337,16 +361,11 @@ impl Command {
            let mut reached_chain_tip = false;
            while accumulated_block_gas < self.target_gas {
                let block_number = next_block;
-                info!(target: "reth-bench", block_number, big_block = big_block_idx, "Fetching block");
+                info!(target: "reth-bench", block_number, big_block = big_block_idx, "Awaiting prefetched block");

-                let fetch_result = tokio::try_join!(
-                    provider.get_block_by_number(block_number.into()).full(),
-                    provider.get_block_receipts(block_number.into()),
-                );
-
-                let (rpc_block, receipts) = match fetch_result {
-                    Ok((Some(block), Some(receipts))) => (block, receipts),
-                    Ok((None, _) | (_, None)) => {
+                let fetched = match block_stream.next().await {
+                    Some(Ok(Some(fetched))) => fetched,
+                    Some(Ok(None)) => {
                        warn!(
                            target: "reth-bench",
                            block_number,
@@ -355,52 +374,16 @@ impl Command {
                        reached_chain_tip = true;
                        break;
                    }
-                    Err(e) => return Err(e.into()),
+                    Some(Err(e)) => return Err(e),
+                    // The block-number stream is open-ended; this only fires if the
+                    // upstream `iter(from..)` is somehow exhausted.
+                    None => {
+                        reached_chain_tip = true;
+                        break;
+                    }
                };
-
-                let block_access_list = if self.bal {
-                    Some(fetch_block_access_list(&provider, block_number).await.wrap_err_with(
-                        || format!("Failed to fetch BAL for block {block_number}"),
-                    )?)
-                } else {
-                    None
-                };
-
-                // Convert RPC receipts to consensus receipts
-                let consensus_receipts: Vec<Receipt> = receipts
-                    .iter()
-                    .map(|r| {
-                        let inner = &r.inner.inner.inner;
-                        let tx_type = r.inner.inner.r#type.try_into().unwrap_or_default();
-                        Receipt {
-                            tx_type,
-                            success: inner.receipt.status.coerce_status(),
-                            cumulative_gas_used: inner.receipt.cumulative_gas_used,
-                            logs: inner
-                                .receipt
-                                .logs
-                                .iter()
-                                .map(|log| alloy_primitives::Log {
-                                    address: log.inner.address,
-                                    data: log.inner.data.clone(),
-                                })
-                                .collect(),
-                        }
-                    })
-                    .collect();
-
-                // Convert to consensus block
-                let block = rpc_block
-                    .into_inner()
-                    .map_header(|header| header.map(|h| h.into_header_with_defaults()))
-                    .try_map_transactions(|tx| -> eyre::Result<TxEnvelope> {
-                        tx.try_into().map_err(|_| eyre::eyre!("unsupported tx type"))
-                    })?
-                    .into_consensus();
-
-                // Convert to ExecutionData
-                let (payload, sidecar) = ExecutionPayload::from_block_slow(&block);
-                let execution_data = ExecutionData { payload, sidecar };
+                let FetchedBlock { execution_data, consensus_receipts, block_access_list } =
+                    fetched;

                let block_gas = execution_data.payload.as_v1().gas_used;
                let block_blob_gas =
@@ -674,6 +657,79 @@ impl Command {
    }
 }

+/// One fully-materialized block fetched by the prefetcher.
+struct FetchedBlock {
+    /// Execution payload with sidecar derived from the RPC block.
+    execution_data: ExecutionData,
+    /// Consensus-format receipts (`cumulative_gas_used` is still per-block, callers offset
+    /// it when merging).
+    consensus_receipts: Vec<Receipt>,
+    /// `eth_getBlockAccessListByBlockNumber` result when `--bal` is enabled.
+    block_access_list: Option<BlockAccessList>,
+}
+
+/// Fetches one block + receipts (and optionally its BAL) from the RPC. Returns `Ok(None)`
+/// when the block doesn't exist yet (chain-tip reached).
+async fn fetch_one_block(
+    provider: RootProvider<AnyNetwork>,
+    block_number: u64,
+    bal_enabled: bool,
+) -> eyre::Result<Option<FetchedBlock>> {
+    let (rpc_block, receipts) = tokio::try_join!(
+        provider.get_block_by_number(block_number.into()).full(),
+        provider.get_block_receipts(block_number.into()),
+    )?;
+    let (rpc_block, receipts) = match (rpc_block, receipts) {
+        (Some(b), Some(r)) => (b, r),
+        _ => return Ok(None),
+    };
+
+    let block_access_list = if bal_enabled {
+        Some(
+            fetch_block_access_list(&provider, block_number)
+                .await
+                .wrap_err_with(|| format!("Failed to fetch BAL for block {block_number}"))?,
+        )
+    } else {
+        None
+    };
+
+    let consensus_receipts: Vec<Receipt> = receipts
+        .iter()
+        .map(|r| {
+            let inner = &r.inner.inner.inner;
+            let tx_type = r.inner.inner.r#type.try_into().unwrap_or_default();
+            Receipt {
+                tx_type,
+                success: inner.receipt.status.coerce_status(),
+                cumulative_gas_used: inner.receipt.cumulative_gas_used,
+                logs: inner
+                    .receipt
+                    .logs
+                    .iter()
+                    .map(|log| alloy_primitives::Log {
+                        address: log.inner.address,
+                        data: log.inner.data.clone(),
+                    })
+                    .collect(),
+            }
+        })
+        .collect();
+
+    let block = rpc_block
+        .into_inner()
+        .map_header(|header| header.map(|h| h.into_header_with_defaults()))
+        .try_map_transactions(|tx| -> eyre::Result<TxEnvelope> {
+            tx.try_into().map_err(|_| eyre::eyre!("unsupported tx type"))
+        })?
+        .into_consensus();
+
+    let (payload, sidecar) = ExecutionPayload::from_block_slow(&block);
+    let execution_data = ExecutionData { payload, sidecar };
+
+    Ok(Some(FetchedBlock { execution_data, consensus_receipts, block_access_list }))
+}
+
 fn merge_block_access_list(
    merged: &mut BlockAccessList,
    incoming: BlockAccessList,
--- a/crates/cli/commands/src/re_execute.rs
+++ b/crates/cli/commands/src/re_execute.rs
@@ -20,7 +20,10 @@ use reth_provider::{
 };
 use reth_revm::{
    database::StateProviderDatabase,
-    db::{states::reverts::AccountInfoRevert, BundleState},
+    db::{
+        states::reverts::{AccountInfoRevert, RevertToSlot},
+        BundleState,
+    },
 };
 use reth_stages::stages::calculate_gas_used_from_headers;
 use reth_storage_api::{ChangeSetReader, DBProvider, StorageChangeSetReader};
@@ -191,10 +194,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                            }
                        };

-                        let bal= executor.take_bal();
-
                        if let Err(err) = consensus
-                            .validate_block_post_execution(&block, &result, None,bal)
+                            .validate_block_post_execution(&block, &result, None)
                            .wrap_err_with(|| {
                                format!(
                                    "Failed to validate block {} {}",
@@ -427,14 +428,19 @@ where
            let mut cs_slots = cs_storage.get_mut(addr);
            for (slot_key, revert_slot) in &revert.storage {
                let b256_key = B256::from(*slot_key);
-                match cs_slots.as_mut().and_then(|s| s.remove(&b256_key)) {
-                    Some(cs_value) => eyre::ensure!(
-                        revert_slot.to_previous_value() == cs_value,
+                let cs_value = cs_slots.as_mut().and_then(|s| s.remove(&b256_key));
+                match (revert_slot, cs_value) {
+                    // When a contract is selfdestructed and re-created at the same address
+                    // within the same block, revm marks slots touched by the new contract
+                    // as `Destroyed` and never reads the original DB value, so
+                    // `to_previous_value()` would resolve to zero, which might be wrong.
+                    (RevertToSlot::Destroyed, _) => {}
+                    (RevertToSlot::Some(prev), Some(cs_value)) => eyre::ensure!(
+                        *prev == cs_value,
                        "Block {block_number}: {addr} slot {b256_key} mismatch: \
-                         revert={} cs={cs_value}",
-                        revert_slot.to_previous_value(),
+                         revert={prev} cs={cs_value}",
                    ),
-                    None => eyre::ensure!(
+                    (RevertToSlot::Some(_), None) => eyre::ensure!(
                        revert.wipe_storage,
                        "Block {block_number}: {addr} slot {b256_key} in reverts but not in changeset",
                    ),
--- a/crates/consensus/consensus/Cargo.toml
+++ b/crates/consensus/consensus/Cargo.toml
@@ -18,7 +18,6 @@ reth-primitives-traits.workspace = true
 # ethereum
 alloy-primitives.workspace = true
 alloy-consensus.workspace = true
-alloy-eip7928.workspace = true

 # misc
 auto_impl.workspace = true
@@ -30,9 +29,10 @@ std = [
    "reth-primitives-traits/std",
    "alloy-primitives/std",
    "alloy-consensus/std",
-    "alloy-eip7928/std",
+    "reth-primitives-traits/std",
    "reth-execution-types/std",
    "thiserror/std",
-    "alloy-eip7928/std",
 ]
-test-utils = ["reth-primitives-traits/test-utils"]
+test-utils = [
+    "reth-primitives-traits/test-utils",
+]
--- a/crates/consensus/consensus/src/lib.rs
+++ b/crates/consensus/consensus/src/lib.rs
@@ -38,7 +38,6 @@ use alloc::{
    vec::Vec,
 };
 use alloy_consensus::Header;
-use alloy_eip7928::BlockAccessList;
 use alloy_primitives::{BlockHash, BlockNumber, Bloom, B256};
 use core::{error::Error, fmt::Display};

@@ -86,7 +85,6 @@ pub trait FullConsensus<N: NodePrimitives>: Consensus<N::Block> {
        block: &RecoveredBlock<N::Block>,
        result: &BlockExecutionResult<N::Receipt>,
        receipt_root_bloom: Option<ReceiptRootBloom>,
-        block_access_list: Option<BlockAccessList>,
    ) -> Result<(), ConsensusError>;
 }

@@ -476,12 +474,6 @@ pub enum ConsensusError {
    /// EIP-7825: Transaction gas limit exceeds maximum allowed
    #[error(transparent)]
    TransactionGasLimitTooHigh(Box<TxGasLimitTooHighErr>),
-    /// Error when an unexpected block access list cost is encountered.
-    #[error("block access list exceeds gas limit")]
-    BlockAccessListExceedsGasLimit,
-    /// Error when the block access list hash doesn't match the expected value.
-    #[error("block access list hash mismatch: {0}")]
-    BlockAccessListHashMismatch(GotExpectedBoxed<B256>),
    /// Any additional consensus error, for example L2-specific errors.
    #[error(transparent)]
    Other(#[from] Arc<dyn Error + Send + Sync>),
@@ -527,23 +519,6 @@ impl ConsensusError {
    }
 }

-/// Validates the block access list against the gas limit.
-///
-/// EIP-7925 specifies that the total cost of the block access list items must not exceed
-/// the gas limit. Each item costs `ITEM_COST` gas.
-pub fn validate_block_access_list_gas(
-    block_access_list: Option<&alloy_eip7928::BlockAccessList>,
-    gas_limit: u64,
-) -> Result<(), ConsensusError> {
-    if let Some(bal) = block_access_list {
-        let bal_items = alloy_eip7928::total_bal_items(bal);
-        if bal_items > gas_limit / alloy_eip7928::ITEM_COST as u64 {
-            return Err(ConsensusError::BlockAccessListExceedsGasLimit)
-        }
-    }
-    Ok(())
-}
-
 impl From<InvalidTransactionError> for ConsensusError {
    fn from(value: InvalidTransactionError) -> Self {
        Self::InvalidTransaction(value)
--- a/crates/consensus/consensus/src/noop.rs
+++ b/crates/consensus/consensus/src/noop.rs
@@ -20,7 +20,6 @@

 use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
 use alloc::sync::Arc;
-use alloy_eip7928::BlockAccessList;
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};

@@ -78,7 +77,6 @@ impl<N: NodePrimitives> FullConsensus<N> for NoopConsensus {
        _block: &RecoveredBlock<N::Block>,
        _result: &BlockExecutionResult<N::Receipt>,
        _receipt_root_bloom: Option<ReceiptRootBloom>,
-        _block_access_list: Option<BlockAccessList>,
    ) -> Result<(), ConsensusError> {
        Ok(())
    }
--- a/crates/consensus/consensus/src/test_utils.rs
+++ b/crates/consensus/consensus/src/test_utils.rs
@@ -1,5 +1,4 @@
 use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
-use alloy_eip7928::BlockAccessList;
 use core::sync::atomic::{AtomicBool, Ordering};
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
@@ -53,7 +52,6 @@ impl<N: NodePrimitives> FullConsensus<N> for TestConsensus {
        _block: &RecoveredBlock<N::Block>,
        _result: &BlockExecutionResult<N::Receipt>,
        _receipt_root_bloom: Option<ReceiptRootBloom>,
-        _block_access_list: Option<BlockAccessList>,
    ) -> Result<(), ConsensusError> {
        if self.fail_validation() {
            Err(ConsensusError::BaseFeeMissing)
--- a/crates/e2e-test-utils/src/payload.rs
+++ b/crates/e2e-test-utils/src/payload.rs
@@ -1,8 +1,8 @@
 use futures_util::StreamExt;
-use reth_node_api::{BlockBody, PayloadAttributes, PayloadKind};
+use reth_node_api::{PayloadAttributes, PayloadKind};
 use reth_payload_builder::{PayloadBuilderHandle, PayloadId};
 use reth_payload_builder_primitives::Events;
-use reth_payload_primitives::{BuiltPayload, PayloadTypes};
+use reth_payload_primitives::PayloadTypes;
 use tokio_stream::wrappers::BroadcastStream;

 /// Helper for payload operations
@@ -53,27 +53,11 @@ impl<T: PayloadTypes> PayloadTestContext<T> {
    ///
    /// Panics if the payload builder does not produce a non-empty payload within 30 seconds.
    pub async fn wait_for_built_payload(&self, payload_id: PayloadId) {
-        let start = std::time::Instant::now();
-        loop {
-            let payload =
-                self.payload_builder.best_payload(payload_id).await.transpose().ok().flatten();
-            if payload.is_none_or(|p| p.block().body().transactions().is_empty()) {
-                assert!(
-                    start.elapsed() < std::time::Duration::from_secs(30),
-                    "timed out waiting for a non-empty payload for {payload_id} — \
-                     check that the chain spec supports all generated tx types"
-                );
-                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
-                continue
-            }
-            // Resolve payload once its built
-            self.payload_builder
-                .resolve_kind(payload_id, PayloadKind::Earliest)
-                .await
-                .unwrap()
-                .unwrap();
-            break;
-        }
+        self.payload_builder
+            .resolve_kind(payload_id, PayloadKind::WaitForPending)
+            .await
+            .unwrap()
+            .unwrap();
    }

    /// Expects the next event to be a built payload event or panics
--- a/crates/engine/primitives/src/config.rs
+++ b/crates/engine/primitives/src/config.rs
@@ -191,9 +191,9 @@ pub struct TreeConfig {
    /// When disabled, the BAL hashed post state is not sent to the multiproof task for
    /// early parallel state root computation.
    disable_bal_parallel_state_root: bool,
-    /// Whether to disable BAL (Block Access List) batched IO during prewarming.
-    /// When disabled, falls back to individual per-slot storage reads instead of
-    /// batched cursor reads via `storage_range`.
+    /// Whether to disable BAL (Block Access List) storage prefetch IO during prewarming.
+    /// When set, BAL storage slots are not read into the execution cache. BAL hashed-state
+    /// streaming for parallel state-root computation is controlled separately.
    disable_bal_batch_io: bool,
    /// Maximum random jitter applied before each proof computation (trie-debug only).
    /// When set, each proof worker sleeps for a random duration up to this value
--- a/crates/engine/primitives/src/forkchoice.rs
+++ b/crates/engine/primitives/src/forkchoice.rs
@@ -21,7 +21,8 @@ impl ForkchoiceStateTracker {
    /// `sync_target` to `None`, since we're now fully synced.
    pub const fn set_latest(&mut self, state: ForkchoiceState, status: ForkchoiceStatus) {
        if status.is_valid() {
-            self.set_valid(state);
+            self.last_syncing = None;
+            self.last_valid = Some(state);
        } else if status.is_syncing() {
            self.last_syncing = Some(state);
        }
@@ -30,11 +31,24 @@ impl ForkchoiceStateTracker {
        self.latest = Some(received);
    }

-    const fn set_valid(&mut self, state: ForkchoiceState) {
-        // we no longer need to sync to this state.
+    /// Promotes a previously tracked syncing forkchoice state to valid, without overwriting a
+    /// newer `latest` state.
+    ///
+    /// This is used when a `Syncing` FCU's head finally becomes canonical via the downloaded-block
+    /// flow, so the safe/finalized anchors of that FCU can be applied. Unlike
+    /// [`Self::set_latest`], this preserves a newer `latest` (e.g. an `Invalid` FCU received
+    /// after the syncing one) and only flips `latest` to `Valid` when it still refers to the same
+    /// syncing FCU being promoted.
+    pub fn promote_sync_target_to_valid(&mut self, state: ForkchoiceState) {
        self.last_syncing = None;
-
        self.last_valid = Some(state);
+
+        if let Some(received) = self.latest.as_mut() &&
+            received.state == state &&
+            received.status.is_syncing()
+        {
+            received.status = ForkchoiceStatus::Valid;
+        }
    }

    /// Returns the [`ForkchoiceStatus`] of the latest received FCU.
--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -1917,9 +1917,37 @@ where
            self.on_canonical_chain_update(chain_update);
        }

+        self.on_canonicalized_sync_target(target);
+
        Ok(())
    }

+    /// Applies the tracked forkchoice state once its sync target head becomes canonical.
+    fn on_canonicalized_sync_target(&mut self, target: B256) {
+        let Some(sync_target_state) = self
+            .state
+            .forkchoice_state_tracker
+            .sync_target_state()
+            .filter(|state| state.head_block_hash == target)
+        else {
+            return;
+        };
+
+        if let Err(outcome) = self.ensure_consistent_forkchoice_state(sync_target_state) {
+            debug!(
+                target: "engine::tree",
+                head = %sync_target_state.head_block_hash,
+                safe = %sync_target_state.safe_block_hash,
+                finalized = %sync_target_state.finalized_block_hash,
+                ?outcome,
+                "Canonicalized sync target head before safe/finalized could be applied"
+            );
+            return;
+        }
+
+        self.state.forkchoice_state_tracker.promote_sync_target_to_valid(sync_target_state);
+    }
+
    /// Convenience function to handle an optional tree event.
    fn on_maybe_tree_event(&mut self, event: Option<TreeEvent>) -> ProviderResult<()> {
        if let Some(event) = event {
--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -123,7 +123,6 @@ where
    /// Whether sparse trie cache pruning is fully disabled.
    disable_sparse_trie_cache_pruning: bool,
    /// Whether to disable BAL-based parallel execution (falls back to tx-based prewarming).
-    #[allow(unused)]
    disable_bal_parallel_execution: bool,
    /// Whether to disable BAL-driven parallel state root computation.
    disable_bal_parallel_state_root: bool,
@@ -273,9 +272,7 @@ where
            halve_workers,
            config,
        );
-        // If no BALs are present or we have them explicitly disabled, we use sparse trie task and
-        // need to send the updates to it via state hook
-        let install_state_hook = env.decoded_bal.is_none() || self.disable_bal_parallel_state_root;
+        let install_state_hook = env.decoded_bal.is_none();
        let prewarm_handle = self.spawn_caching_with(
            env,
            prewarm_rx,
@@ -508,14 +505,14 @@ where
        );
        {
            let to_prewarm_task = to_prewarm_task.clone();
-            let disable_bal_parallel_state_root = self.disable_bal_parallel_state_root;
+            let disable_bal_parallel_execution = self.disable_bal_parallel_execution;
            self.executor.spawn_blocking_named("prewarm", move || {
-                let mode = if let Some(decoded_bal) =
-                    maybe_decoded_bal.filter(|_| !disable_bal_parallel_state_root)
+                let mode = if skip_prewarm {
+                    PrewarmMode::Skipped
+                } else if let Some(decoded_bal) =
+                    maybe_decoded_bal.filter(|_| !disable_bal_parallel_execution)
                {
                    PrewarmMode::BlockAccessList(decoded_bal)
-                } else if skip_prewarm {
-                    PrewarmMode::Skipped
                } else {
                    PrewarmMode::Transactions(transactions)
                };
@@ -801,7 +798,7 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {

    /// Returns a state hook to stream execution state updates to the sparse trie cache task.
    ///
-    /// Returns `None` when BAL-driven hashed state streaming feeds the sparse trie task.
+    /// Returns `None` when execution should not send state updates, such as BAL-driven execution.
    pub fn state_hook(&self) -> Option<impl OnStateHook> {
        self.install_state_hook
            .then(|| self.state_root_handle.as_ref().map(|handle| handle.state_hook()))
@@ -1162,16 +1159,19 @@ mod tests {
                    }
                }

-                let mut account = revm_state::Account::default();
-                account.info = AccountInfo {
-                    balance: U256::from(rng.random::<u64>()),
-                    nonce: rng.random::<u64>(),
-                    code_hash: KECCAK_EMPTY,
-                    code: Some(Default::default()),
-                    account_id: None,
+                let account = revm_state::Account {
+                    info: AccountInfo {
+                        balance: U256::from(rng.random::<u64>()),
+                        nonce: rng.random::<u64>(),
+                        code_hash: KECCAK_EMPTY,
+                        code: Some(Default::default()),
+                        account_id: None,
+                    },
+                    original_info: Box::new(AccountInfo::default()),
+                    storage,
+                    status: AccountStatus::Touched,
+                    transaction_id: 0,
                };
-                account.storage = storage;
-                account.status = AccountStatus::Touched;

                state_update.insert(address, account);
            }
--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -62,9 +62,7 @@ use crate::tree::payload_processor::receipt_root_task::{IndexedReceipt, ReceiptR
 use reth_chain_state::{
    CanonicalInMemoryState, DeferredTrieData, ExecutedBlock, ExecutionTimingStats, LazyOverlay,
 };
-use reth_consensus::{
-    validate_block_access_list_gas, ConsensusError, FullConsensus, ReceiptRootBloom,
-};
+use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom};
 use reth_engine_primitives::{
    ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator,
 };
@@ -570,7 +568,7 @@ where
        // The receipt root task is spawned before execution and receives receipts incrementally
        // as transactions complete, allowing parallel computation during execution.
        let execute_block_start = Instant::now();
-        let (output, senders, receipt_root_rx, built_bal) =
+        let (output, senders, receipt_root_rx) =
            match self.execute_block(state_provider, env, &input, &mut handle) {
                Ok(output) => output,
                Err(err) => return self.handle_execution_error(input, err, &parent_block),
@@ -652,7 +650,6 @@ where
                transaction_root,
                receipt_root_bloom,
                hashed_state,
-                built_bal
            ),
            block
        );
@@ -909,7 +906,6 @@ where
            BlockExecutionOutput<N::Receipt>,
            Vec<Address>,
            tokio::sync::oneshot::Receiver<(B256, alloy_primitives::Bloom)>,
-            Option<BlockAccessList>,
        ),
        InsertBlockErrorKind,
    >
@@ -917,29 +913,15 @@ where
        S: StateProvider + Send,
        Err: core::error::Error + Send + Sync + 'static,
        V: PayloadValidator<T, Block = N::Block>,
-        T: PayloadTypes<
-            BuiltPayload: BuiltPayload<Primitives = N>,
-            ExecutionData: ExecutionPayload,
-        >,
+        T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
        Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
    {
        debug!(target: "engine::tree::payload_validator", "Executing block");

-        if let Some(bal_opt) = input.block_access_list() {
-            let bal = bal_opt.map_err(BlockExecutionError::other)?;
-            validate_block_access_list_gas(Some(&bal), input.gas_limit())
-                .map_err(|e| {
-                    debug!(target: "engine::tree::payload_validator", "BAL is invalid since it contains more items than the gas limit allows");
-                    InsertBlockErrorKind::Consensus(e)
-                })?
-        }
-
-        let has_bal = input.block_access_list().is_some();
        let mut db = debug_span!(target: "engine::tree", "build_state_db").in_scope(|| {
            State::builder()
                .with_database(StateProviderDatabase::new(state_provider))
                .with_bundle_update()
-                .with_bal_builder_if(has_bal)
                .build()
        });

@@ -998,7 +980,6 @@ where
            handle.iter_transactions(),
            &receipt_tx,
            &executed_tx_index,
-            has_bal,
        )?;
        drop(receipt_tx);

@@ -1013,11 +994,6 @@ where
        debug_span!(target: "engine::tree", "merge_transitions")
            .in_scope(|| db.merge_transitions(BundleRetention::Reverts));

-        // Extract the built bal if payload has bal
-        let built_bal = if has_bal { db.take_built_alloy_bal() } else { None };
-
-        tracing::debug!(has_bal = built_bal.is_some(), "Built BAL");
-
        let output = BlockExecutionOutput { result, state: db.take_bundle() };

        let execution_duration = execution_start.elapsed();
@@ -1025,7 +1001,7 @@ where
        self.metrics.record_block_execution_gas_bucket(output.result.gas_used, execution_duration);
        debug!(target: "engine::tree::payload_validator", elapsed = ?execution_duration, "Executed block");

-        Ok((output, senders, result_rx, built_bal))
+        Ok((output, senders, result_rx))
    }

    /// Executes transactions and collects senders, streaming receipts to a background task.
@@ -1037,20 +1013,18 @@ where
    /// - Collecting transaction senders for later use
    ///
    /// Returns the executor (for finalization) and the collected senders.
-    fn execute_transactions<'a, E, Tx, InnerTx, Err, DB>(
+    fn execute_transactions<E, Tx, InnerTx, Err>(
        &self,
        mut executor: E,
        transaction_count: usize,
        transactions: impl Iterator<Item = Result<Tx, Err>>,
        receipt_tx: &crossbeam_channel::Sender<IndexedReceipt<N::Receipt>>,
        executed_tx_index: &AtomicUsize,
-        has_bal: bool,
    ) -> Result<(E, Vec<Address>), BlockExecutionError>
    where
-        E: BlockExecutor<Receipt = N::Receipt, Evm: alloy_evm::Evm<DB = &'a mut State<DB>>>,
+        E: BlockExecutor<Receipt = N::Receipt>,
        Tx: alloy_evm::block::ExecutableTx<E> + alloy_evm::RecoveredTx<InnerTx>,
        InnerTx: TxHashRef,
-        DB: revm::Database + 'a,
        Err: core::error::Error + Send + Sync + 'static,
    {
        let mut senders = Vec::with_capacity(transaction_count);
@@ -1061,11 +1035,6 @@ where
            .in_scope(|| executor.apply_pre_execution_changes())?;
        self.metrics.record_pre_execution(pre_exec_start.elapsed());

-        // Bump BAL index after pre-execution changes (EIP-7928: index 0 is pre-execution)
-        if has_bal {
-            executor.evm_mut().db_mut().bump_bal_index();
-        }
-
        // Execute transactions
        let exec_span = debug_span!(target: "engine::tree", "execution").entered();
        let mut transactions = transactions.into_iter();
@@ -1110,10 +1079,6 @@ where
                    let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
                }
            }
-            // Bump BAL index after each transaction (EIP-7928)
-            if has_bal {
-                executor.evm_mut().db_mut().bump_bal_index();
-            }
        }
        drop(exec_span);

@@ -1397,7 +1362,6 @@ where
        transaction_root: Option<B256>,
        receipt_root_bloom: Option<ReceiptRootBloom>,
        hashed_state: LazyHashedPostState,
-        built_bal: Option<BlockAccessList>,
    ) -> Result<LazyHashedPostState, InsertBlockErrorKind>
    where
        V: PayloadValidator<T, Block = N::Block>,
@@ -1424,13 +1388,9 @@ where
        let _enter =
            debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution")
                .entered();
-
-        if let Err(err) = self.consensus.validate_block_post_execution(
-            block,
-            output,
-            receipt_root_bloom,
-            built_bal,
-        ) {
+        if let Err(err) =
+            self.consensus.validate_block_post_execution(block, output, receipt_root_bloom)
+        {
            // call post-block hook
            self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
            return Err(err.into())
--- a/crates/engine/tree/src/tree/precompile_cache.rs
+++ b/crates/engine/tree/src/tree/precompile_cache.rs
@@ -13,7 +13,7 @@ use std::{hash::Hash, sync::Arc};
 use tracing::error;

 /// Default max cache size for [`PrecompileCache`]
-const MAX_CACHE_SIZE: u32 = 10_000;
+const MAX_CACHE_SIZE: u32 = 1024 * 1024;

 /// Stores caches for each precompile.
 #[derive(Debug, Clone, Default)]
@@ -54,6 +54,9 @@ where
            moka::sync::CacheBuilder::new(MAX_CACHE_SIZE as u64)
                .initial_capacity(MAX_CACHE_SIZE as usize)
                .eviction_policy(EvictionPolicy::lru())
+                .weigher(|key: &Bytes, value: &CacheEntry<S>| {
+                    (key.len() + value.output.bytes.len()) as u32
+                })
                .build_with_hasher(Default::default()),
        )
    }
@@ -266,7 +269,6 @@ mod tests {
                state_gas_used: 0,
                reservoir: 0,
                gas_refunded: 0,
-                refill_amount: 0,
                bytes: Bytes::default(),
            })
        })
@@ -281,7 +283,6 @@ mod tests {
            state_gas_used: 0,
            reservoir: 0,
            gas_refunded: 0,
-            refill_amount: 0,
            bytes: alloy_primitives::Bytes::copy_from_slice(b"cached_result"),
        };

@@ -316,7 +317,6 @@ mod tests {
                    state_gas_used: 0,
                    reservoir: 0,
                    gas_refunded: 0,
-                    refill_amount: 0,
                    bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_1"),
                })
            }
@@ -334,7 +334,6 @@ mod tests {
                    state_gas_used: 0,
                    reservoir: 0,
                    gas_refunded: 0,
-                    refill_amount: 0,
                    bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_2"),
                })
            }
--- a/crates/engine/tree/src/tree/tests.rs
+++ b/crates/engine/tree/src/tree/tests.rs
@@ -2254,3 +2254,65 @@ fn test_on_valid_downloaded_head_sync_target_returns_make_canonical() {
        other => panic!("Expected MakeCanonical for head block, got: {other:?}"),
    }
 }
+
+/// Tests that canonicalizing a downloaded sync target head also applies the tracked finalized
+/// block from the original `SYNCING` forkchoice state.
+#[test]
+fn test_canonicalizing_downloaded_sync_target_head_updates_finalized() {
+    reth_tracing::init_test_tracing();
+
+    let chain_spec = MAINNET.clone();
+    let mut test_harness = TestHarness::new(chain_spec);
+
+    let blocks: Vec<_> = test_harness.block_builder.get_executed_blocks(0..3).collect();
+    let genesis = &blocks[0];
+    let finalized_block = &blocks[1];
+    let head_block = &blocks[2];
+
+    test_harness = test_harness.with_blocks(vec![
+        genesis.clone(),
+        finalized_block.clone(),
+        head_block.clone(),
+    ]);
+
+    let finalized_num_hash = finalized_block.recovered_block().num_hash();
+    let head_num_hash = head_block.recovered_block().num_hash();
+
+    test_harness.tree.state.tree_state.set_canonical_head(genesis.recovered_block().num_hash());
+
+    let fcu_state = ForkchoiceState {
+        head_block_hash: head_num_hash.hash,
+        safe_block_hash: head_num_hash.hash,
+        finalized_block_hash: finalized_num_hash.hash,
+    };
+    test_harness
+        .tree
+        .state
+        .forkchoice_state_tracker
+        .set_latest(fcu_state, ForkchoiceStatus::Syncing);
+
+    let event = test_harness
+        .tree
+        .on_valid_downloaded_block(head_num_hash)
+        .unwrap()
+        .expect("expected canonicalization event for sync target head");
+
+    test_harness.tree.on_tree_event(event).unwrap();
+
+    assert_eq!(test_harness.tree.state.tree_state.canonical_block_hash(), head_num_hash.hash);
+    assert_eq!(
+        test_harness.tree.canonical_in_memory_state.get_finalized_num_hash(),
+        Some(finalized_num_hash),
+        "Finalized block from the syncing FCU should be applied once the head becomes canonical"
+    );
+    assert_eq!(
+        test_harness.tree.canonical_in_memory_state.get_safe_num_hash(),
+        Some(head_num_hash),
+        "Safe block from the syncing FCU should be applied once the head becomes canonical"
+    );
+    assert_eq!(
+        test_harness.tree.state.forkchoice_state_tracker.last_valid_state(),
+        Some(fcu_state)
+    );
+    assert!(test_harness.tree.state.forkchoice_state_tracker.sync_target_state().is_none());
+}
--- a/crates/ethereum/consensus/src/lib.rs
+++ b/crates/ethereum/consensus/src/lib.rs
@@ -13,7 +13,7 @@ extern crate alloc;

 use alloc::{fmt::Debug, sync::Arc};
 use alloy_consensus::{constants::MAXIMUM_EXTRA_DATA_SIZE, EMPTY_OMMER_ROOT_HASH};
-use alloy_eips::{eip7840::BlobParams, eip7928::BlockAccessList};
+use alloy_eips::eip7840::BlobParams;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_consensus::{
    Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom, TransactionRoot,
@@ -108,15 +108,9 @@ where
        block: &RecoveredBlock<N::Block>,
        result: &BlockExecutionResult<N::Receipt>,
        receipt_root_bloom: Option<ReceiptRootBloom>,
-        block_access_list: Option<BlockAccessList>,
    ) -> Result<(), ConsensusError> {
-        let res = validate_block_post_execution(
-            block,
-            &self.chain_spec,
-            result,
-            receipt_root_bloom,
-            block_access_list,
-        );
+        let res =
+            validate_block_post_execution(block, &self.chain_spec, result, receipt_root_bloom);

        if self.skip_requests_hash_check &&
            let Err(ConsensusError::BodyRequestsHashDiff(_)) = &res
--- a/crates/ethereum/consensus/src/validation.rs
+++ b/crates/ethereum/consensus/src/validation.rs
@@ -1,9 +1,6 @@
 use alloc::vec::Vec;
 use alloy_consensus::{proofs::calculate_receipt_root, BlockHeader, TxReceipt};
-use alloy_eips::{
-    eip7928::{compute_block_access_list_hash, BlockAccessList},
-    Encodable2718,
-};
+use alloy_eips::Encodable2718;
 use alloy_primitives::{Bloom, Bytes, B256};
 use reth_chainspec::EthereumHardforks;
 use reth_consensus::ConsensusError;
@@ -24,7 +21,6 @@ pub fn validate_block_post_execution<B, R, ChainSpec>(
    chain_spec: &ChainSpec,
    result: &BlockExecutionResult<R>,
    receipt_root_bloom: Option<(B256, Bloom)>,
-    block_access_list: Option<BlockAccessList>,
 ) -> Result<(), ConsensusError>
 where
    B: Block,
@@ -83,21 +79,6 @@ where
        }
    }

-    // Validate that the block access list hash matches the calculated block access list hash
-    if chain_spec.is_amsterdam_active_at_timestamp(block.header().timestamp()) &&
-        block_access_list.is_some()
-    {
-        let block_bal_hash = block.header().block_access_list_hash().unwrap_or_default();
-        let default_bal = BlockAccessList::default();
-        let block_access_list_hash =
-            compute_block_access_list_hash(block_access_list.as_ref().unwrap_or(&default_bal));
-        if block_access_list_hash != block_bal_hash {
-            return Err(ConsensusError::BlockAccessListHashMismatch(
-                (block_access_list_hash, block_bal_hash).into(),
-            ))
-        }
-    }
-
    Ok(())
 }

--- a/crates/ethereum/evm/src/build.rs
+++ b/crates/ethereum/evm/src/build.rs
@@ -47,7 +47,6 @@ where
            transactions,
            output: BlockExecutionResult { receipts, requests, gas_used, blob_gas_used },
            state_root,
-            block_access_list_hash,
            ..
        } = input;

@@ -91,12 +90,6 @@ where
            };
        }

-        let bal_hash = if self.chain_spec.is_amsterdam_active_at_timestamp(timestamp) {
-            block_access_list_hash
-        } else {
-            None
-        };
-
        let header = Header {
            parent_hash: ctx.parent_hash,
            ommers_hash: EMPTY_OMMER_ROOT_HASH,
@@ -119,8 +112,8 @@ where
            blob_gas_used: block_blob_gas_used,
            excess_blob_gas,
            requests_hash,
-            block_access_list_hash: bal_hash,
-            slot_number: ctx.slot_number,
+            block_access_list_hash: None,
+            slot_number: None,
        };

        Ok(Block {
--- a/crates/ethereum/payload/src/lib.rs
+++ b/crates/ethereum/payload/src/lib.rs
@@ -9,7 +9,7 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]

 use alloy_consensus::Transaction;
-use alloy_primitives::{Bytes, U256};
+use alloy_primitives::U256;
 use alloy_rlp::Encodable;
 use alloy_rpc_types_engine::PayloadAttributes as EthPayloadAttributes;
 use reth_basic_payload_builder::{
@@ -446,9 +446,7 @@ where
        return Ok(BuildOutcome::Aborted { fees: total_fees, cached_reads })
    }

-    let BlockBuilderOutcome { execution_result, block, block_access_list, .. } = if let Some(
-        mut handle,
-    ) = trie_handle
+    let BlockBuilderOutcome { execution_result, block, .. } = if let Some(mut handle) = trie_handle
    {
        // Drop the state hook, which drops the StateHookSender and triggers
        // FinishedStateUpdates via its Drop impl, signaling the trie task to finalize.
@@ -487,10 +485,8 @@ where
            max_rlp_length: MAX_RLP_BLOCK_SIZE,
        }));
    }
-    let block_access_list: Option<Bytes> =
-        block_access_list.map(|block_access_list| alloy_rlp::encode(&block_access_list).into());

-    let payload = EthBuiltPayload::new(sealed_block, total_fees, requests, block_access_list)
+    let payload = EthBuiltPayload::new(sealed_block, total_fees, requests, None)
        // add blob sidecars from the executed txs
        .with_sidecars(blob_sidecars);

--- a/crates/evm/evm/src/either.rs
+++ b/crates/evm/evm/src/either.rs
@@ -79,11 +79,4 @@ where
            Self::Right(b) => b.size_hint(),
        }
    }
-
-    fn take_bal(&mut self) -> Option<alloy_eips::eip7928::BlockAccessList> {
-        match self {
-            Self::Left(a) => a.take_bal(),
-            Self::Right(b) => b.take_bal(),
-        }
-    }
 }
--- a/crates/evm/evm/src/execute.rs
+++ b/crates/evm/evm/src/execute.rs
@@ -3,10 +3,7 @@
 use crate::{ConfigureEvm, Database, OnStateHook, TxEnvFor};
 use alloc::{boxed::Box, sync::Arc, vec::Vec};
 use alloy_consensus::{BlockHeader, Header};
-use alloy_eips::{
-    eip2718::WithEncoded,
-    eip7928::{compute_block_access_list_hash, BlockAccessList},
-};
+use alloy_eips::eip2718::WithEncoded;
 pub use alloy_evm::block::{BlockExecutor, BlockExecutorFactory, GasOutput};
 use alloy_evm::{
    block::{CommitChanges, ExecutableTxParts},
@@ -24,10 +21,7 @@ use reth_primitives_traits::{
 use reth_storage_api::StateProvider;
 pub use reth_storage_errors::provider::ProviderError;
 use reth_trie_common::{updates::TrieUpdates, HashedPostState};
-use revm::{
-    database::{states::bundle_state::BundleRetention, BundleState, State},
-    state::bal::Bal,
-};
+use revm::database::{states::bundle_state::BundleRetention, BundleState, State};

 /// A type that knows how to execute a block. It is assumed to operate on a
 /// [`crate::Evm`] internally and use [`State`] as database.
@@ -151,9 +145,6 @@ pub trait Executor<DB: Database>: Sized {
    ///
    /// This is used to optimize DB commits depending on the size of the state.
    fn size_hint(&self) -> usize;
-
-    /// Take built [`BlockAccessList`] from executor
-    fn take_bal(&mut self) -> Option<BlockAccessList>;
 }

 /// Input for block building. Consumed by [`BlockAssembler`].
@@ -171,7 +162,6 @@ pub trait Executor<DB: Database>: Sized {
 /// - `bundle_state`: Accumulated state changes from all transactions
 /// - `state_provider`: Access to the current state for additional lookups
 /// - `state_root`: The calculated state root after all changes
-/// - `block_access_list_hash`: Block access list hash (EIP-7928, Amsterdam)
 ///
 /// # Usage
 ///
@@ -188,7 +178,6 @@ pub trait Executor<DB: Database>: Sized {
 ///     bundle_state: &state_changes,
 ///     state_provider: &state,
 ///     state_root: calculated_root,
-///     block_access_list_hash: Some(calculated_bal_hash),
 /// };
 ///
 /// let block = assembler.assemble_block(input)?;
@@ -216,8 +205,6 @@ pub struct BlockAssemblerInput<'a, 'b, F: BlockExecutorFactory, H = Header> {
    pub state_provider: &'b dyn StateProvider,
    /// State root for this block.
    pub state_root: B256,
-    /// Block access list hash (EIP-7928, Amsterdam).
-    pub block_access_list_hash: Option<B256>,
 }

 impl<'a, 'b, F: BlockExecutorFactory, H> BlockAssemblerInput<'a, 'b, F, H> {
@@ -235,7 +222,6 @@ impl<'a, 'b, F: BlockExecutorFactory, H> BlockAssemblerInput<'a, 'b, F, H> {
        bundle_state: &'a BundleState,
        state_provider: &'b dyn StateProvider,
        state_root: B256,
-        block_access_list_hash: Option<B256>,
    ) -> Self {
        Self {
            evm_env,
@@ -246,7 +232,6 @@ impl<'a, 'b, F: BlockExecutorFactory, H> BlockAssemblerInput<'a, 'b, F, H> {
            bundle_state,
            state_provider,
            state_root,
-            block_access_list_hash,
        }
    }
 }
@@ -316,8 +301,6 @@ pub struct BlockBuilderOutcome<N: NodePrimitives> {
    pub trie_updates: TrieUpdates,
    /// The built block.
    pub block: RecoveredBlock<N::Block>,
-    /// Block access list built during execution (EIP-7928, Amsterdam).
-    pub block_access_list: Option<BlockAccessList>,
 }

 /// A type that knows how to execute and build a block.
@@ -470,11 +453,7 @@ where
    type Executor = Executor;

    fn apply_pre_execution_changes(&mut self) -> Result<(), BlockExecutionError> {
-        self.executor.apply_pre_execution_changes()?;
-        // Bump BAL index after pre-execution changes (EIP-7928: index 0 is pre-execution)
-        self.executor.evm_mut().db_mut().bump_bal_index();
-
-        Ok(())
+        self.executor.apply_pre_execution_changes()
    }

    fn execute_transaction_with_commit_condition(
@@ -487,8 +466,6 @@ where
            self.executor.execute_transaction_with_commit_condition((tx_env, &tx), f)?
        {
            self.transactions.push(tx);
-            // Bump BAL index after each committed transaction (EIP-7928)
-            self.executor.evm_mut().db_mut().bump_bal_index();
            Ok(Some(gas_used))
        } else {
            Ok(None)
@@ -506,11 +483,6 @@ where
        // merge all transitions into bundle state
        db.merge_transitions(BundleRetention::Reverts);

-        // extract the built block access list (EIP-7928, Amsterdam) and compute its hash
-        let block_access_list = db.take_built_alloy_bal();
-        let block_access_list_hash =
-            block_access_list.as_ref().map(|bal| compute_block_access_list_hash(bal));
-
        let hashed_state = state.hashed_post_state(&db.bundle_state);
        let (state_root, trie_updates) = match state_root_precomputed {
            Some(precomputed) => precomputed,
@@ -531,18 +503,11 @@ where
            bundle_state: &db.bundle_state,
            state_provider: &state,
            state_root,
-            block_access_list_hash,
        })?;

        let block = RecoveredBlock::new_unhashed(block, senders);

-        Ok(BlockBuilderOutcome {
-            execution_result: result,
-            hashed_state,
-            trie_updates,
-            block,
-            block_access_list,
-        })
+        Ok(BlockBuilderOutcome { execution_result: result, hashed_state, trie_updates, block })
    }

    fn executor_mut(&mut self) -> &mut Self::Executor {
@@ -589,33 +554,11 @@ where
        block: &RecoveredBlock<<Self::Primitives as NodePrimitives>::Block>,
    ) -> Result<BlockExecutionResult<<Self::Primitives as NodePrimitives>::Receipt>, Self::Error>
    {
-        let mut executor = self
+        let result = self
            .strategy_factory
            .executor_for_block(&mut self.db, block)
-            .map_err(BlockExecutionError::other)?;
-
-        let has_bal = block.header().block_access_list_hash().is_some();
-
-        if has_bal {
-            executor.evm_mut().db_mut().bal_state.bal_builder = Some(Bal::new());
-        } else {
-            executor.evm_mut().db_mut().bal_state.bal_builder = None;
-        }
-
-        executor.apply_pre_execution_changes()?;
-
-        if has_bal {
-            executor.evm_mut().db_mut().bump_bal_index();
-        }
-
-        for tx in block.transactions_recovered() {
-            executor.execute_transaction(tx)?;
-            if has_bal {
-                executor.evm_mut().db_mut().bump_bal_index();
-            }
-        }
-
-        let result = executor.apply_post_execution_changes()?;
+            .map_err(BlockExecutionError::other)?
+            .execute_block(block.transactions_recovered())?;

        self.db.merge_transitions(BundleRetention::Reverts);

@@ -649,10 +592,6 @@ where
    fn size_hint(&self) -> usize {
        self.db.bundle_state.size_hint()
    }
-
-    fn take_bal(&mut self) -> Option<BlockAccessList> {
-        self.db.take_built_alloy_bal()
-    }
 }

 /// A helper trait marking a 'static type that can be converted into an [`ExecutableTxParts`] for
@@ -758,10 +697,6 @@ mod tests {
        fn size_hint(&self) -> usize {
            0
        }
-
-        fn take_bal(&mut self) -> Option<BlockAccessList> {
-            None
-        }
    }

    #[test]
--- a/crates/net/eth-wire-types/src/version.rs
+++ b/crates/net/eth-wire-types/src/version.rs
@@ -35,11 +35,10 @@ pub enum EthVersion {

 impl EthVersion {
    /// The latest known eth version
-    pub const LATEST: Self = Self::Eth70;
+    pub const LATEST: Self = Self::Eth69;

    /// All known eth versions
-    pub const ALL_VERSIONS: &'static [Self] =
-        &[Self::Eth70, Self::Eth69, Self::Eth68, Self::Eth67, Self::Eth66];
+    pub const ALL_VERSIONS: &'static [Self] = &[Self::Eth69, Self::Eth68, Self::Eth67, Self::Eth66];

    /// Returns true if the version is eth/66
    pub const fn is_eth66(&self) -> bool {
--- a/crates/node/core/src/args/engine.rs
+++ b/crates/node/core/src/args/engine.rs
@@ -521,8 +521,8 @@ pub struct EngineArgs {
    #[arg(long = "engine.disable-bal-parallel-state-root", default_value_t = DefaultEngineValues::get_global().bal_parallel_state_root_disabled)]
    pub bal_parallel_state_root_disabled: bool,

-    /// Disable BAL (Block Access List) batched IO during prewarming. When set, falls back
-    /// to individual per-slot storage reads instead of batched cursor reads.
+    /// Disable BAL (Block Access List) storage prefetch IO during prewarming. When set, BAL
+    /// storage slots are not read into the execution cache.
    #[arg(long = "engine.disable-bal-batch-io", default_value_t = false)]
    pub disable_bal_batch_io: bool,

--- a/crates/node/core/src/args/mod.rs
+++ b/crates/node/core/src/args/mod.rs
@@ -2,7 +2,7 @@

 /// NetworkArg struct for configuring the network
 mod network;
-pub use network::{DefaultNetworkArgs, DiscoveryArgs, NetworkArgs};
+pub use network::{DefaultDiscoveryArgs, DefaultNetworkArgs, DiscoveryArgs, NetworkArgs};

 /// RpcServerArg struct for configuring the RPC
 mod rpc_server;
--- a/crates/node/core/src/args/network.rs
+++ b/crates/node/core/src/args/network.rs
@@ -11,7 +11,10 @@ use std::{
 };

 use crate::version::version_metadata;
-use clap::Args;
+use clap::{
+    builder::{OsStr, Resettable},
+    Args,
+};
 use reth_chainspec::EthChainSpec;
 use reth_cli_util::{get_secret_key, load_secret_key::SecretKeyError};
 use reth_config::Config;
@@ -569,15 +572,8 @@ impl NetworkArgs {
                let rlpx_socket = (addr, self.port).into();
                self.discovery.apply_to_builder(builder, rlpx_socket, chain_bootnodes)
            })
-            .listener_addr(SocketAddr::new(
-                addr, // set discovery port based on instance number
-                self.port,
-            ))
-            .discovery_addr(SocketAddr::new(
-                self.discovery.addr,
-                // set discovery port based on instance number
-                self.discovery.port,
-            ))
+            .listener_addr(SocketAddr::new(addr, self.port))
+            .discovery_addr(SocketAddr::new(self.discovery.addr, self.discovery.port))
            .disable_tx_gossip(self.disable_tx_gossip)
            .required_block_hashes(self.required_block_hashes.clone())
            .eth_max_message_size_opt(self.eth_max_message_size.map(NonZeroUsize::get))
@@ -727,19 +723,172 @@ impl Default for NetworkArgs {
    }
 }

+/// Global static discovery defaults
+static DISCOVERY_DEFAULTS: OnceLock<DefaultDiscoveryArgs> = OnceLock::new();
+
+/// Default values for discovery CLI arguments that can be customized.
+#[derive(Debug, Clone, Copy)]
+pub struct DefaultDiscoveryArgs {
+    /// Default for `--disable-discovery`.
+    pub disable_discovery: bool,
+    /// Default for `--disable-dns-discovery`.
+    pub disable_dns_discovery: bool,
+    /// Default for `--disable-discv4-discovery`.
+    pub disable_discv4_discovery: bool,
+    /// Default for `--disable-discv5-discovery`.
+    pub disable_discv5_discovery: bool,
+    /// Default for `--disable-nat`.
+    pub disable_nat: bool,
+    /// Default UDP address for devp2p discovery v4.
+    pub addr: IpAddr,
+    /// Default UDP port for devp2p discovery v4.
+    pub port: u16,
+    /// Default UDP IPv4 address for devp2p discovery v5.
+    pub discv5_addr: Option<Ipv4Addr>,
+    /// Default UDP IPv6 address for devp2p discovery v5.
+    pub discv5_addr_ipv6: Option<Ipv6Addr>,
+    /// Default UDP IPv4 port for devp2p discovery v5.
+    pub discv5_port: Option<u16>,
+    /// Default UDP IPv6 port for devp2p discovery v5.
+    pub discv5_port_ipv6: Option<u16>,
+    /// Default discv5 periodic lookup interval (seconds).
+    pub discv5_lookup_interval: u64,
+    /// Default discv5 bootstrap lookup interval (seconds).
+    pub discv5_bootstrap_lookup_interval: u64,
+    /// Default discv5 bootstrap lookup countdown.
+    pub discv5_bootstrap_lookup_countdown: u64,
+}
+
+impl DefaultDiscoveryArgs {
+    /// Initialize the global discovery defaults with this configuration.
+    pub fn try_init(self) -> Result<(), Self> {
+        DISCOVERY_DEFAULTS.set(self)
+    }
+
+    /// Get a reference to the global discovery defaults.
+    pub fn get_global() -> &'static Self {
+        DISCOVERY_DEFAULTS.get_or_init(Self::default)
+    }
+
+    /// Set the default for `--disable-discovery`.
+    pub const fn with_disable_discovery(mut self, disable: bool) -> Self {
+        self.disable_discovery = disable;
+        self
+    }
+
+    /// Set the default for `--disable-dns-discovery`.
+    pub const fn with_disable_dns_discovery(mut self, disable: bool) -> Self {
+        self.disable_dns_discovery = disable;
+        self
+    }
+
+    /// Set the default for `--disable-discv4-discovery`.
+    pub const fn with_disable_discv4_discovery(mut self, disable: bool) -> Self {
+        self.disable_discv4_discovery = disable;
+        self
+    }
+
+    /// Set the default for `--disable-discv5-discovery`.
+    pub const fn with_disable_discv5_discovery(mut self, disable: bool) -> Self {
+        self.disable_discv5_discovery = disable;
+        self
+    }
+
+    /// Set the default for `--disable-nat`.
+    pub const fn with_disable_nat(mut self, disable: bool) -> Self {
+        self.disable_nat = disable;
+        self
+    }
+
+    /// Set the default discovery v4 address.
+    pub const fn with_addr(mut self, addr: IpAddr) -> Self {
+        self.addr = addr;
+        self
+    }
+
+    /// Set the default discovery v4 port.
+    pub const fn with_port(mut self, port: u16) -> Self {
+        self.port = port;
+        self
+    }
+
+    /// Set the default discovery v5 IPv4 address.
+    pub fn with_discv5_addr(mut self, addr: impl Into<Option<Ipv4Addr>>) -> Self {
+        self.discv5_addr = addr.into();
+        self
+    }
+
+    /// Set the default discovery v5 IPv6 address.
+    pub fn with_discv5_addr_ipv6(mut self, addr: impl Into<Option<Ipv6Addr>>) -> Self {
+        self.discv5_addr_ipv6 = addr.into();
+        self
+    }
+
+    /// Set the default discovery V5 port.
+    pub fn with_discv5_port(mut self, port: impl Into<Option<u16>>) -> Self {
+        self.discv5_port = port.into();
+        self
+    }
+
+    /// Set the default discovery v5 IPv6 port.
+    pub fn with_discv5_port_ipv6(mut self, port: impl Into<Option<u16>>) -> Self {
+        self.discv5_port_ipv6 = port.into();
+        self
+    }
+
+    /// Set the default discv5 periodic lookup interval (seconds).
+    pub const fn with_discv5_lookup_interval(mut self, interval: u64) -> Self {
+        self.discv5_lookup_interval = interval;
+        self
+    }
+
+    /// Set the default discv5 bootstrap lookup interval (seconds).
+    pub const fn with_discv5_bootstrap_lookup_interval(mut self, interval: u64) -> Self {
+        self.discv5_bootstrap_lookup_interval = interval;
+        self
+    }
+
+    /// Set the default discv5 bootstrap lookup countdown.
+    pub const fn with_discv5_bootstrap_lookup_countdown(mut self, countdown: u64) -> Self {
+        self.discv5_bootstrap_lookup_countdown = countdown;
+        self
+    }
+}
+
+impl Default for DefaultDiscoveryArgs {
+    fn default() -> Self {
+        Self {
+            disable_discovery: false,
+            disable_dns_discovery: false,
+            disable_discv4_discovery: false,
+            disable_discv5_discovery: false,
+            disable_nat: false,
+            addr: DEFAULT_DISCOVERY_ADDR,
+            port: DEFAULT_DISCOVERY_PORT,
+            discv5_addr: None,
+            discv5_addr_ipv6: None,
+            discv5_port: Some(DEFAULT_DISCOVERY_V5_PORT),
+            discv5_port_ipv6: Some(DEFAULT_DISCOVERY_V5_PORT),
+            discv5_lookup_interval: DEFAULT_SECONDS_LOOKUP_INTERVAL,
+            discv5_bootstrap_lookup_interval: DEFAULT_SECONDS_BOOTSTRAP_LOOKUP_INTERVAL,
+            discv5_bootstrap_lookup_countdown: DEFAULT_COUNT_BOOTSTRAP_LOOKUPS,
+        }
+    }
+}
+
 /// Arguments to setup discovery
 #[derive(Debug, Clone, Args, PartialEq, Eq)]
 pub struct DiscoveryArgs {
    /// Disable the discovery service.
-    #[arg(short, long, default_value_if("dev", "true", "true"))]
+    #[arg(short, long, default_value_if("dev", "true", "true"), default_value_t = DefaultDiscoveryArgs::get_global().disable_discovery)]
    pub disable_discovery: bool,

    /// Disable the DNS discovery.
-    #[arg(long, conflicts_with = "disable_discovery")]
+    #[arg(long, conflicts_with = "disable_discovery", default_value_t = DefaultDiscoveryArgs::get_global().disable_dns_discovery)]
    pub disable_dns_discovery: bool,

    /// Disable Discv4 discovery.
-    #[arg(long, conflicts_with = "disable_discovery")]
+    #[arg(long, conflicts_with = "disable_discovery", default_value_t = DefaultDiscoveryArgs::get_global().disable_discv4_discovery)]
    pub disable_discv4_discovery: bool,

    /// Enable Discv5 discovery.
@@ -750,57 +899,57 @@ pub struct DiscoveryArgs {
    pub enable_discv5_discovery: bool,

    /// Disable Discv5 discovery.
-    #[arg(long, conflicts_with = "disable_discovery")]
+    #[arg(long, conflicts_with = "disable_discovery", default_value_t = DefaultDiscoveryArgs::get_global().disable_discv5_discovery)]
    pub disable_discv5_discovery: bool,

    /// Disable Nat discovery.
-    #[arg(long, conflicts_with = "disable_discovery")]
+    #[arg(long, conflicts_with = "disable_discovery", default_value_t = DefaultDiscoveryArgs::get_global().disable_nat)]
    pub disable_nat: bool,

    /// The UDP address to use for devp2p peer discovery version 4.
-    #[arg(id = "discovery.addr", long = "discovery.addr", value_name = "DISCOVERY_ADDR", default_value_t = DEFAULT_DISCOVERY_ADDR)]
+    #[arg(id = "discovery.addr", long = "discovery.addr", value_name = "DISCOVERY_ADDR", default_value_t = DefaultDiscoveryArgs::get_global().addr)]
    pub addr: IpAddr,

    /// The UDP port to use for devp2p peer discovery version 4.
-    #[arg(id = "discovery.port", long = "discovery.port", value_name = "DISCOVERY_PORT", default_value_t = DEFAULT_DISCOVERY_PORT)]
+    #[arg(id = "discovery.port", long = "discovery.port", value_name = "DISCOVERY_PORT", default_value_t = DefaultDiscoveryArgs::get_global().port)]
    pub port: u16,

    /// The UDP IPv4 address to use for devp2p peer discovery version 5. Overwritten by `RLPx`
    /// address, if it's also IPv4.
-    #[arg(id = "discovery.v5.addr", long = "discovery.v5.addr", value_name = "DISCOVERY_V5_ADDR", default_value = None)]
+    #[arg(id = "discovery.v5.addr", long = "discovery.v5.addr", value_name = "DISCOVERY_V5_ADDR", default_value = Resettable::from(DefaultDiscoveryArgs::get_global().discv5_addr.map(|a| OsStr::from(a.to_string()))))]
    pub discv5_addr: Option<Ipv4Addr>,

    /// The UDP IPv6 address to use for devp2p peer discovery version 5. Overwritten by `RLPx`
    /// address, if it's also IPv6.
-    #[arg(id = "discovery.v5.addr.ipv6", long = "discovery.v5.addr.ipv6", value_name = "DISCOVERY_V5_ADDR_IPV6", default_value = None)]
+    #[arg(id = "discovery.v5.addr.ipv6", long = "discovery.v5.addr.ipv6", value_name = "DISCOVERY_V5_ADDR_IPV6", default_value = Resettable::from(DefaultDiscoveryArgs::get_global().discv5_addr_ipv6.map(|a| OsStr::from(a.to_string()))))]
    pub discv5_addr_ipv6: Option<Ipv6Addr>,

    /// The UDP IPv4 port to use for devp2p peer discovery version 5. Not used unless `--addr` is
    /// IPv4, or `--discovery.v5.addr` is set.
-    #[arg(id = "discovery.v5.port", long = "discovery.v5.port", value_name = "DISCOVERY_V5_PORT",
-    default_value_t = DEFAULT_DISCOVERY_V5_PORT)]
-    pub discv5_port: u16,
+    #[arg(id = "discovery.v5.port", long = "discovery.v5.port", value_name = "DISCOVERY_V5_PORT", default_value = Resettable::from(DefaultDiscoveryArgs::get_global().discv5_port.map(|p| OsStr::from(p.to_string()))))]
+    pub discv5_port: Option<u16>,

    /// The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is
    /// IPv6, or `--discovery.addr.ipv6` is set.
-    #[arg(id = "discovery.v5.port.ipv6", long = "discovery.v5.port.ipv6", value_name = "DISCOVERY_V5_PORT_IPV6",
-    default_value_t = DEFAULT_DISCOVERY_V5_PORT)]
-    pub discv5_port_ipv6: u16,
+    ///
+    /// If not provided, discovery V5 defaults to same port as discovery V4 (--discovery.port).
+    #[arg(id = "discovery.v5.port.ipv6", long = "discovery.v5.port.ipv6", value_name = "DISCOVERY_V5_PORT_IPV6", default_value = Resettable::from(DefaultDiscoveryArgs::get_global().discv5_port_ipv6.map(|p| OsStr::from(p.to_string()))))]
+    pub discv5_port_ipv6: Option<u16>,

    /// The interval in seconds at which to carry out periodic lookup queries, for the whole
    /// run of the program.
-    #[arg(id = "discovery.v5.lookup-interval", long = "discovery.v5.lookup-interval", value_name = "DISCOVERY_V5_LOOKUP_INTERVAL", default_value_t = DEFAULT_SECONDS_LOOKUP_INTERVAL)]
+    #[arg(id = "discovery.v5.lookup-interval", long = "discovery.v5.lookup-interval", value_name = "DISCOVERY_V5_LOOKUP_INTERVAL", default_value_t = DefaultDiscoveryArgs::get_global().discv5_lookup_interval)]
    pub discv5_lookup_interval: u64,

    /// The interval in seconds at which to carry out boost lookup queries, for a fixed number of
    /// times, at bootstrap.
    #[arg(id = "discovery.v5.bootstrap.lookup-interval", long = "discovery.v5.bootstrap.lookup-interval", value_name = "DISCOVERY_V5_BOOTSTRAP_LOOKUP_INTERVAL",
-        default_value_t = DEFAULT_SECONDS_BOOTSTRAP_LOOKUP_INTERVAL)]
+        default_value_t = DefaultDiscoveryArgs::get_global().discv5_bootstrap_lookup_interval)]
    pub discv5_bootstrap_lookup_interval: u64,

    /// The number of times to carry out boost lookup queries at bootstrap.
    #[arg(id = "discovery.v5.bootstrap.lookup-countdown", long = "discovery.v5.bootstrap.lookup-countdown", value_name = "DISCOVERY_V5_BOOTSTRAP_LOOKUP_COUNTDOWN",
-        default_value_t = DEFAULT_COUNT_BOOTSTRAP_LOOKUPS)]
+        default_value_t = DefaultDiscoveryArgs::get_global().discv5_bootstrap_lookup_countdown)]
    pub discv5_bootstrap_lookup_countdown: u64,
 }

@@ -850,6 +999,7 @@ impl DiscoveryArgs {
            discv5_lookup_interval,
            discv5_bootstrap_lookup_interval,
            discv5_bootstrap_lookup_countdown,
+            port,
            ..
        } = self;

@@ -867,8 +1017,9 @@ impl DiscoveryArgs {

        let mut discv5_config_builder =
            reth_discv5::discv5::ConfigBuilder::new(ListenConfig::from_two_sockets(
-                discv5_addr_ipv4.map(|addr| SocketAddrV4::new(addr, *discv5_port)),
-                discv5_addr_ipv6.map(|addr| SocketAddrV6::new(addr, *discv5_port_ipv6, 0, 0)),
+                discv5_addr_ipv4.map(|addr| SocketAddrV4::new(addr, discv5_port.unwrap_or(*port))),
+                discv5_addr_ipv6
+                    .map(|addr| SocketAddrV6::new(addr, discv5_port_ipv6.unwrap_or(*port), 0, 0)),
            ));

        if has_discv5_addr_args || self.disable_nat {
@@ -898,14 +1049,14 @@ impl DiscoveryArgs {
    /// discovery binds to the sockets.
    pub const fn with_unused_discovery_port(mut self) -> Self {
        self.port = 0;
-        self.discv5_port = 0;
-        self.discv5_port_ipv6 = 0;
+        self.discv5_port = Some(0);
+        self.discv5_port_ipv6 = Some(0);
        self
    }

    /// Set the discovery V5 port
-    pub const fn with_discv5_port(mut self, port: u16) -> Self {
-        self.discv5_port = port;
+    pub fn with_discv5_port(mut self, port: impl Into<Option<u16>>) -> Self {
+        self.discv5_port = port.into();
        self
    }

@@ -917,29 +1068,45 @@ impl DiscoveryArgs {
    pub fn adjust_instance_ports(&mut self, instance: u16) {
        debug_assert_ne!(instance, 0, "instance must be non-zero");
        self.port += instance - 1;
-        self.discv5_port += instance - 1;
-        self.discv5_port_ipv6 += instance - 1;
+        self.discv5_port = self.discv5_port.map(|port| port + instance - 1);
+        self.discv5_port_ipv6 = self.discv5_port_ipv6.map(|port| port + instance - 1);
    }
 }

 impl Default for DiscoveryArgs {
    fn default() -> Self {
+        let DefaultDiscoveryArgs {
+            disable_discovery,
+            disable_dns_discovery,
+            disable_discv4_discovery,
+            disable_discv5_discovery,
+            disable_nat,
+            addr,
+            port,
+            discv5_addr,
+            discv5_addr_ipv6,
+            discv5_port,
+            discv5_port_ipv6,
+            discv5_lookup_interval,
+            discv5_bootstrap_lookup_interval,
+            discv5_bootstrap_lookup_countdown,
+        } = *DefaultDiscoveryArgs::get_global();
        Self {
-            disable_discovery: false,
-            disable_dns_discovery: false,
-            disable_discv4_discovery: false,
+            disable_discovery,
+            disable_dns_discovery,
+            disable_discv4_discovery,
            enable_discv5_discovery: false,
-            disable_discv5_discovery: false,
-            disable_nat: false,
-            addr: DEFAULT_DISCOVERY_ADDR,
-            port: DEFAULT_DISCOVERY_PORT,
-            discv5_addr: None,
-            discv5_addr_ipv6: None,
-            discv5_port: DEFAULT_DISCOVERY_V5_PORT,
-            discv5_port_ipv6: DEFAULT_DISCOVERY_V5_PORT,
-            discv5_lookup_interval: DEFAULT_SECONDS_LOOKUP_INTERVAL,
-            discv5_bootstrap_lookup_interval: DEFAULT_SECONDS_BOOTSTRAP_LOOKUP_INTERVAL,
-            discv5_bootstrap_lookup_countdown: DEFAULT_COUNT_BOOTSTRAP_LOOKUPS,
+            disable_discv5_discovery,
+            disable_nat,
+            addr,
+            port,
+            discv5_addr,
+            discv5_addr_ipv6,
+            discv5_port,
+            discv5_port_ipv6,
+            discv5_lookup_interval,
+            discv5_bootstrap_lookup_interval,
+            discv5_bootstrap_lookup_countdown,
        }
    }
 }
--- a/crates/rpc/rpc-api/src/engine.rs
+++ b/crates/rpc/rpc-api/src/engine.rs
@@ -4,12 +4,12 @@
 //! the consensus client.

 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip7685::RequestsOrHash,
    BlockId, BlockNumberOrTag,
 };
 use alloy_json_rpc::RpcObject;
-use alloy_primitives::{Address, BlockHash, Bytes, B256, U256, U64};
+use alloy_primitives::{Address, BlockHash, Bytes, B128, B256, U256, U64};
 use alloy_rpc_types_engine::{
    ClientVersionV1, ExecutionPayloadBodiesV1, ExecutionPayloadBodiesV2, ExecutionPayloadInputV2,
    ExecutionPayloadV1, ExecutionPayloadV3, ExecutionPayloadV4, ForkchoiceState, ForkchoiceUpdated,
@@ -324,6 +324,20 @@ pub trait EngineApi<Engine: EngineTypes> {
        &self,
        versioned_hashes: Vec<B256>,
    ) -> RpcResult<Option<Vec<Option<BlobAndProofV2>>>>;
+
+    /// Fetch blob cells for the consensus layer from the blob store.
+    ///
+    /// Returns a response of the same length as the request. Missing blobs are returned as `null`
+    /// elements; missing requested cells within an available blob are returned as `null` cell and
+    /// proof entries.
+    ///
+    /// Returns `null` if syncing.
+    #[method(name = "getBlobsV4")]
+    async fn get_blobs_v4(
+        &self,
+        versioned_hashes: Vec<B256>,
+        indices_bitarray: B128,
+    ) -> RpcResult<Option<Vec<Option<BlobCellsAndProofsV1>>>>;
 }

 /// A subset of the ETH rpc interface: <https://ethereum.github.io/execution-apis/api-documentation>
--- a/crates/rpc/rpc-engine-api/src/capabilities.rs
+++ b/crates/rpc/rpc-engine-api/src/capabilities.rs
@@ -37,6 +37,7 @@ pub const CAPABILITIES: &[&str] = &[
    "engine_getBlobsV1",
    "engine_getBlobsV2",
    "engine_getBlobsV3",
+    "engine_getBlobsV4",
 ];

 /// Engine API capabilities set.
@@ -218,6 +219,7 @@ mod tests {

        assert!(!is_critical_method("engine_getBlobsV1"));
        assert!(!is_critical_method("engine_getBlobsV3"));
+        assert!(!is_critical_method("engine_getBlobsV4"));
        assert!(!is_critical_method("engine_getPayloadBodiesByHashV1"));
        assert!(!is_critical_method("engine_getPayloadBodiesByRangeV1"));
        assert!(!is_critical_method("engine_getClientVersionV1"));
--- a/crates/rpc/rpc-engine-api/src/engine_api.rs
+++ b/crates/rpc/rpc-engine-api/src/engine_api.rs
@@ -3,11 +3,11 @@ use crate::{
 };
 use alloy_eips::{
    eip1898::BlockHashOrNumber,
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip4895::Withdrawals,
    eip7685::RequestsOrHash,
 };
-use alloy_primitives::{BlockHash, BlockNumber, B256, U64};
+use alloy_primitives::{BlockHash, BlockNumber, B128, B256, U64};
 use alloy_rpc_types_engine::{
    CancunPayloadFields, ClientVersionV1, ExecutionData, ExecutionPayloadBodiesV1,
    ExecutionPayloadBodiesV2, ExecutionPayloadBodyV1, ExecutionPayloadBodyV2,
@@ -953,6 +953,35 @@ where
            .map_err(|err| EngineApiError::Internal(Box::new(err)))
    }

+    fn get_blobs_v4(
+        &self,
+        versioned_hashes: Vec<B256>,
+        indices_bitarray: B128,
+    ) -> EngineApiResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
+        let current_timestamp =
+            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap_or_default().as_secs();
+        if !self.inner.chain_spec.is_amsterdam_active_at_timestamp(current_timestamp) {
+            return Err(EngineApiError::EngineObjectValidationError(
+                reth_payload_primitives::EngineObjectValidationError::UnsupportedFork,
+            ));
+        }
+
+        if versioned_hashes.len() > MAX_BLOB_LIMIT {
+            return Err(EngineApiError::BlobRequestTooLarge { len: versioned_hashes.len() })
+        }
+
+        // Spec requires returning `null` if syncing.
+        if (*self.inner.is_syncing)() {
+            return Ok(None)
+        }
+
+        self.inner
+            .tx_pool
+            .get_blobs_for_versioned_hashes_v4(&versioned_hashes, indices_bitarray)
+            .map(Some)
+            .map_err(|err| EngineApiError::Internal(Box::new(err)))
+    }
+
    /// Metered version of `get_blobs_v2`.
    pub fn get_blobs_v2_metered(
        &self,
@@ -1009,6 +1038,28 @@ where

        res
    }
+
+    /// Metered version of `get_blobs_v4`.
+    pub fn get_blobs_v4_metered(
+        &self,
+        versioned_hashes: Vec<B256>,
+        indices_bitarray: B128,
+    ) -> EngineApiResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
+        let hashes_len = versioned_hashes.len();
+        let start = Instant::now();
+        let res = Self::get_blobs_v4(self, versioned_hashes, indices_bitarray);
+        self.inner.metrics.latency.get_blobs_v4.record(start.elapsed());
+
+        if let Ok(Some(blobs)) = &res {
+            let blobs_found = blobs.iter().flatten().count();
+            let blobs_missed = hashes_len - blobs_found;
+
+            self.inner.metrics.blob_metrics.blob_count.increment(blobs_found as u64);
+            self.inner.metrics.blob_metrics.blob_misses.increment(blobs_missed as u64);
+        }
+
+        res
+    }
 }

 // This is the concrete ethereum engine API implementation.
@@ -1375,6 +1426,15 @@ where
        trace!(target: "rpc::engine", "Serving engine_getBlobsV3");
        Ok(self.get_blobs_v3_metered(versioned_hashes)?)
    }
+
+    async fn get_blobs_v4(
+        &self,
+        versioned_hashes: Vec<B256>,
+        indices_bitarray: B128,
+    ) -> RpcResult<Option<Vec<Option<BlobCellsAndProofsV1>>>> {
+        trace!(target: "rpc::engine", "Serving engine_getBlobsV4");
+        Ok(self.get_blobs_v4_metered(versioned_hashes, indices_bitarray)?)
+    }
 }

 impl<Provider, EngineT, Pool, Validator, ChainSpec> IntoEngineApiRpcModule
@@ -1660,6 +1720,37 @@ mod tests {
        assert_matches!(res, Ok(None));
    }

+    #[tokio::test]
+    async fn get_blobs_v4_returns_null_when_syncing() {
+        let chain_spec: Arc<ChainSpec> =
+            Arc::new(ChainSpecBuilder::mainnet().amsterdam_activated().build());
+        let provider = Arc::new(MockEthProvider::default());
+        let payload_store = spawn_test_payload_service::<EthEngineTypes>();
+        let (to_engine, _engine_rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();
+
+        let api = EngineApi::new(
+            provider,
+            chain_spec.clone(),
+            ConsensusEngineHandle::new(to_engine),
+            payload_store.into(),
+            NoopTransactionPool::default(),
+            Runtime::test(),
+            ClientVersionV1 {
+                code: ClientCode::RH,
+                name: "Reth".to_string(),
+                version: "v0.0.0-test".to_string(),
+                commit: "test".to_string(),
+            },
+            EngineCapabilities::default(),
+            EthereumEngineValidator::new(chain_spec),
+            false,
+            TestNetworkInfo { syncing: true },
+        );
+
+        let res = api.get_blobs_v4_metered(vec![B256::ZERO], B128::from(1u128));
+        assert_matches!(res, Ok(None));
+    }
+
    #[tokio::test]
    async fn fcu_v3_syncing_precedes_invalid_payload_attributes_validation() {
        let (mut handle, api) = setup_engine_api();
--- a/crates/rpc/rpc-engine-api/src/metrics.rs
+++ b/crates/rpc/rpc-engine-api/src/metrics.rs
@@ -58,6 +58,8 @@ pub(crate) struct EngineApiLatencyMetrics {
    pub(crate) get_blobs_v2: Histogram,
    /// Latency for `engine_getBlobsV3`
    pub(crate) get_blobs_v3: Histogram,
+    /// Latency for `engine_getBlobsV4`
+    pub(crate) get_blobs_v4: Histogram,
 }

 #[derive(Metrics)]
--- a/crates/rpc/rpc/src/eth/filter.rs
+++ b/crates/rpc/rpc/src/eth/filter.rs
@@ -737,18 +737,21 @@ where
                is_multi_block_range &&
                all_logs.len() > max_logs_per_response
            {
+                let retry_to_block =
+                    if num_hash.number == from_block { from_block } else { num_hash.number - 1 };
+
                debug!(
                    target: "rpc::eth::filter",
                    logs_found = all_logs.len(),
                    max_logs_per_response,
                    from_block,
-                    to_block = num_hash.number,
+                    to_block = retry_to_block,
                    "Query exceeded max logs per response limit"
                );
                return Err(EthFilterError::QueryExceedsMaxResults {
                    max_logs: max_logs_per_response,
                    from_block,
-                    to_block: num_hash.number,
+                    to_block: retry_to_block,
                });
            }
        }
@@ -1816,6 +1819,87 @@ mod tests {
        assert!(result.is_none());
    }

+    #[tokio::test]
+    async fn test_log_limit_retry_range_excludes_overflow_block() {
+        let provider = MockEthProvider::default();
+
+        use alloy_consensus::TxLegacy;
+        use reth_db_api::models::StoredBlockBodyIndices;
+        use reth_ethereum_primitives::{TransactionSigned, TxType};
+
+        let tx_inner = TxLegacy {
+            chain_id: Some(1),
+            nonce: 0,
+            gas_price: 21_000,
+            gas_limit: 21_000,
+            to: alloy_primitives::TxKind::Call(alloy_primitives::Address::ZERO),
+            value: alloy_primitives::U256::ZERO,
+            input: alloy_primitives::Bytes::new(),
+        };
+        let signature = alloy_primitives::Signature::test_signature();
+        let tx = TransactionSigned::new_unhashed(tx_inner.into(), signature);
+
+        let mock_log = alloy_primitives::Log {
+            address: alloy_primitives::Address::ZERO,
+            data: alloy_primitives::LogData::new_unchecked(vec![], alloy_primitives::Bytes::new()),
+        };
+
+        let receipt = reth_ethereum_primitives::Receipt {
+            tx_type: TxType::Legacy,
+            cumulative_gas_used: 21_000,
+            logs: vec![mock_log],
+            success: true,
+        };
+
+        let mut prev_hash = alloy_primitives::B256::default();
+        for (idx, block_number) in (100u64..=102).enumerate() {
+            let header = alloy_consensus::Header {
+                number: block_number,
+                parent_hash: prev_hash,
+                logs_bloom: alloy_primitives::Bloom::from([1u8; 256]),
+                ..Default::default()
+            };
+            let hash = header.hash_slow();
+            prev_hash = hash;
+
+            let block = reth_ethereum_primitives::Block {
+                header,
+                body: reth_ethereum_primitives::BlockBody {
+                    transactions: vec![tx.clone()],
+                    ..Default::default()
+                },
+            };
+            provider.add_block(hash, block);
+            provider.add_receipts(block_number, vec![receipt.clone()]);
+            provider.add_block_body_indices(
+                block_number,
+                StoredBlockBodyIndices { first_tx_num: idx as u64, tx_count: 1 },
+            );
+        }
+
+        let eth_api = build_test_eth_api(provider);
+        let eth_filter = EthFilter::new(eth_api, EthFilterConfig::default(), Runtime::test());
+        let err = eth_filter
+            .inner
+            .clone()
+            .get_logs_in_block_range(
+                Filter::default(),
+                100,
+                102,
+                QueryLimits { max_blocks_per_filter: None, max_logs_per_response: Some(2) },
+            )
+            .await
+            .expect_err("range should exceed max logs");
+
+        let EthFilterError::QueryExceedsMaxResults { max_logs, from_block, to_block } = err else {
+            panic!("unexpected error: {err:?}");
+        };
+
+        assert_eq!(max_logs, 2);
+        assert_eq!(from_block, 100);
+        assert_eq!(to_block, 101);
+    }
+
    #[tokio::test]
    async fn test_non_consecutive_headers_after_bloom_filter() {
        let provider = MockEthProvider::default();
--- a/crates/rpc/rpc/src/validation.rs
+++ b/crates/rpc/rpc/src/validation.rs
@@ -202,7 +202,7 @@ where
        // update the cached reads
        self.update_cached_reads(parent_header_hash, request_cache).await;

-        self.consensus.validate_block_post_execution(&block, &output, None, None)?;
+        self.consensus.validate_block_post_execution(&block, &output, None)?;

        self.ensure_payment(&block, &output, &message)?;

--- a/crates/stages/api/src/pipeline/mod.rs
+++ b/crates/stages/api/src/pipeline/mod.rs
@@ -9,7 +9,7 @@ use reth_primitives_traits::constants::BEACON_CONSENSUS_REORG_UNWIND_DEPTH;
 use reth_provider::{
    providers::ProviderNodeTypes, BlockHashReader, BlockNumReader, ChainStateBlockReader,
    ChainStateBlockWriter, DBProvider, DatabaseProviderFactory, ProviderFactory,
-    PruneCheckpointReader, StageCheckpointReader, StageCheckpointWriter,
+    PruneCheckpointReader, StageCheckpointReader, StageCheckpointWriter, StorageSettingsCache,
 };
 use reth_prune::PrunerBuilder;
 use reth_static_file::StaticFileProducer;
@@ -269,9 +269,16 @@ impl<N: ProviderNodeTypes> Pipeline<N> {
    /// - [`StaticFileSegment::Transactions`](reth_static_file_types::StaticFileSegment::Transactions)
    ///   -> [`StageId::Bodies`]
    ///
+    /// This is a legacy storage.v1 backfill step. Storage.v2 writes directly to static files and
+    /// `RocksDB`, so there is no MDBX -> static-file migration to perform.
+    ///
    /// CAUTION: This method locks the static file producer Mutex, hence can block the thread if the
    /// lock is occupied.
    pub fn move_to_static_files(&self) -> RethResult<()> {
+        if self.provider_factory.cached_storage_settings().is_v2() {
+            return Ok(())
+        }
+
        // Copies data from database to static files
        let lowest_static_file_height =
            self.static_file_producer.lock().copy_to_static_files()?.min_block_num();
--- a/crates/stages/stages/src/stages/execution/mod.rs
+++ b/crates/stages/stages/src/stages/execution/mod.rs
@@ -4,7 +4,7 @@ use alloy_primitives::BlockNumber;
 use num_traits::Zero;
 use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
 use reth_config::config::ExecutionConfig;
-use reth_consensus::{validate_block_access_list_gas, FullConsensus};
+use reth_consensus::FullConsensus;
 use reth_db::{static_file::HeaderMask, tables};
 use reth_evm::{execute::Executor, metrics::ExecutorMetrics, ConfigureEvm};
 use reth_execution_types::Chain;
@@ -357,19 +357,7 @@ where
                })
            })?;

-            let bal = executor.take_bal().unwrap_or_default();
-            if block.header().block_access_list_hash().is_some() &&
-                let Err(err) = validate_block_access_list_gas(Some(&bal), block.gas_limit())
-            {
-                return Err(StageError::Block {
-                    block: Box::new(block.block_with_parent()),
-                    error: BlockErrorKind::Validation(err),
-                })
-            }
-
-            if let Err(err) =
-                self.consensus.validate_block_post_execution(&block, &result, None, Some(bal))
-            {
+            if let Err(err) = self.consensus.validate_block_post_execution(&block, &result, None) {
                return Err(StageError::Block {
                    block: Box::new(block.block_with_parent()),
                    error: BlockErrorKind::Validation(err),
--- a/crates/storage/db/src/implementation/mdbx/cursor.rs
+++ b/crates/storage/db/src/implementation/mdbx/cursor.rs
@@ -2,7 +2,7 @@

 use super::utils::*;
 use crate::{
-    metrics::{DatabaseEnvMetrics, Operation},
+    metrics::{Operation, TableOperationMetrics},
    DatabaseError,
 };
 use reth_db_api::{
@@ -15,7 +15,7 @@ use reth_db_api::{
 };
 use reth_libmdbx::{Error as MDBXError, TransactionKind, WriteFlags, RO, RW};
 use reth_storage_errors::db::{DatabaseErrorInfo, DatabaseWriteError, DatabaseWriteOperation};
-use std::{borrow::Cow, collections::Bound, marker::PhantomData, ops::RangeBounds, sync::Arc};
+use std::{borrow::Cow, collections::Bound, marker::PhantomData, ops::RangeBounds};

 /// Read only Cursor.
 pub type CursorRO<T> = Cursor<RO, T>;
@@ -29,8 +29,8 @@ pub struct Cursor<K: TransactionKind, T: Table> {
    pub(crate) inner: reth_libmdbx::Cursor<K>,
    /// Cache buffer that receives compressed values.
    buf: Vec<u8>,
-    /// Reference to metric handles in the DB environment. If `None`, metrics are not recorded.
-    metrics: Option<Arc<DatabaseEnvMetrics>>,
+    /// Per-table operation metrics. If `None`, metrics are not recorded.
+    metrics: Option<TableOperationMetrics>,
    /// Phantom data to enforce encoding/decoding.
    _dbi: PhantomData<T>,
 }
@@ -38,7 +38,7 @@ pub struct Cursor<K: TransactionKind, T: Table> {
 impl<K: TransactionKind, T: Table> Cursor<K, T> {
    pub(crate) const fn new_with_metrics(
        inner: reth_libmdbx::Cursor<K>,
-        metrics: Option<Arc<DatabaseEnvMetrics>>,
+        metrics: Option<TableOperationMetrics>,
    ) -> Self {
        Self { inner, buf: Vec::new(), metrics, _dbi: PhantomData }
    }
@@ -54,7 +54,7 @@ impl<K: TransactionKind, T: Table> Cursor<K, T> {
        f: impl FnOnce(&mut Self) -> R,
    ) -> R {
        if let Some(metrics) = self.metrics.clone() {
-            metrics.record_operation(T::NAME, operation, value_size, || f(self))
+            metrics[operation.index()].record(value_size, || f(self))
        } else {
            f(self)
        }
--- a/crates/storage/db/src/implementation/mdbx/tx.rs
+++ b/crates/storage/db/src/implementation/mdbx/tx.rs
@@ -100,7 +100,7 @@ impl<K: TransactionKind> Tx<K> {

        Ok(Cursor::new_with_metrics(
            inner,
-            self.metrics_handler.as_ref().map(|h| h.env_metrics.clone()),
+            self.metrics_handler.as_ref().map(|h| h.env_metrics.table_operation_metrics(T::NAME)),
        ))
    }

--- a/crates/storage/db/src/metrics.rs
+++ b/crates/storage/db/src/metrics.rs
@@ -3,7 +3,7 @@ use metrics::Histogram;
 use quanta::Instant;
 use reth_metrics::{metrics::Counter, Metrics};
 use rustc_hash::FxHashMap;
-use std::time::Duration;
+use std::{array, sync::Arc, time::Duration};
 use strum::{EnumCount, EnumIter, IntoEnumIterator};

 const LARGE_VALUE_THRESHOLD_BYTES: usize = 4096;
@@ -15,8 +15,8 @@ const LARGE_VALUE_THRESHOLD_BYTES: usize = 4096;
 /// Otherwise, metric recording will no-op.
 #[derive(Debug)]
 pub(crate) struct DatabaseEnvMetrics {
-    /// Caches `OperationMetrics` handles for each table and operation tuple.
-    operations: FxHashMap<(&'static str, Operation), OperationMetrics>,
+    /// Caches per-table operation metric handles for all database operation metrics.
+    operations: FxHashMap<&'static str, TableOperationMetrics>,
    /// Caches `TransactionMetrics` handles for counters grouped by only transaction mode.
    /// Updated both at tx open and close.
    transactions: FxHashMap<TransactionMode, TransactionMetrics>,
@@ -26,6 +26,9 @@ pub(crate) struct DatabaseEnvMetrics {
        FxHashMap<(TransactionMode, TransactionOutcome), TransactionOutcomeMetrics>,
 }

+/// Per-table operation metric handles cached for hot cursor paths.
+pub(crate) type TableOperationMetrics = Arc<[OperationMetrics; Operation::COUNT]>;
+
 impl DatabaseEnvMetrics {
    pub(crate) fn new() -> Self {
        // Pre-populate metric handle maps with all possible combinations of labels
@@ -37,24 +40,23 @@ impl DatabaseEnvMetrics {
        }
    }

-    /// Generate a map of all possible operation handles for each table and operation tuple.
-    /// Used for tracking all operation metrics.
-    fn generate_operation_handles() -> FxHashMap<(&'static str, Operation), OperationMetrics> {
-        let mut operations = FxHashMap::with_capacity_and_hasher(
-            Tables::COUNT * Operation::COUNT,
-            Default::default(),
-        );
+    /// Generate a map of pre-bound operation handles for each table.
+    fn generate_operation_handles() -> FxHashMap<&'static str, TableOperationMetrics> {
+        let mut operations = FxHashMap::with_capacity_and_hasher(Tables::COUNT, Default::default());
+
        for table in Tables::ALL {
-            for operation in Operation::iter() {
-                operations.insert(
-                    (table.name(), operation),
-                    OperationMetrics::new_with_labels(&[
-                        (Labels::Table.as_str(), table.name()),
-                        (Labels::Operation.as_str(), operation.as_str()),
-                    ]),
-                );
-            }
+            let table_name = table.name();
+            let metrics = array::from_fn(|index| {
+                let operation = Operation::from_index(index);
+                OperationMetrics::new_with_labels(&[
+                    (Labels::Table.as_str(), table_name),
+                    (Labels::Operation.as_str(), operation.as_str()),
+                ])
+            });
+
+            operations.insert(table_name, Arc::new(metrics));
        }
+
        operations
    }

@@ -105,13 +107,18 @@ impl DatabaseEnvMetrics {
        value_size: Option<usize>,
        f: impl FnOnce() -> R,
    ) -> R {
-        if let Some(metrics) = self.operations.get(&(table, operation)) {
-            metrics.record(value_size, f)
+        if let Some(metrics) = self.operations.get(table) {
+            metrics[operation.index()].record(value_size, f)
        } else {
            f()
        }
    }

+    /// Returns pre-bound operation metric handles for a single table.
+    pub(crate) fn table_operation_metrics(&self, table: &'static str) -> TableOperationMetrics {
+        self.operations.get(table).expect("table operation metric handles not found").clone()
+    }
+
    /// Record metrics for opening a database transaction.
    pub(crate) fn record_opened_transaction(&self, mode: TransactionMode) {
        self.transactions
@@ -219,6 +226,39 @@ pub(crate) enum Operation {
 }

 impl Operation {
+    /// Returns the index of the operation in the cached per-table operation array.
+    pub(crate) const fn index(&self) -> usize {
+        match self {
+            Self::Get => 0,
+            Self::PutUpsert => 1,
+            Self::PutAppend => 2,
+            Self::Delete => 3,
+            Self::CursorUpsert => 4,
+            Self::CursorInsert => 5,
+            Self::CursorAppend => 6,
+            Self::CursorAppendDup => 7,
+            Self::CursorDeleteCurrent => 8,
+            Self::CursorDeleteCurrentDuplicates => 9,
+        }
+    }
+
+    /// Returns the operation for the given index in the cached per-table operation array.
+    const fn from_index(index: usize) -> Self {
+        match index {
+            0 => Self::Get,
+            1 => Self::PutUpsert,
+            2 => Self::PutAppend,
+            3 => Self::Delete,
+            4 => Self::CursorUpsert,
+            5 => Self::CursorInsert,
+            6 => Self::CursorAppend,
+            7 => Self::CursorAppendDup,
+            8 => Self::CursorDeleteCurrent,
+            9 => Self::CursorDeleteCurrentDuplicates,
+            _ => panic!("invalid operation index"),
+        }
+    }
+
    /// Returns the operation as a string.
    pub(crate) const fn as_str(&self) -> &'static str {
        match self {
--- a/crates/storage/libmdbx-rs/mdbx-sys/build.rs
+++ b/crates/storage/libmdbx-rs/mdbx-sys/build.rs
@@ -97,6 +97,7 @@ fn generate_bindings(mdbx: &Path, out_file: &Path) {
    let bindings = bindgen::Builder::default()
        .header(mdbx.join("mdbx.h").to_string_lossy())
        .allowlist_var("^(MDBX|mdbx)_.*")
+        .blocklist_item("MDBX_NOTLS")
        .allowlist_type("^(MDBX|mdbx)_.*")
        .allowlist_function("^(MDBX|mdbx)_.*")
        .size_t_is_usize(true)
--- a/crates/storage/libmdbx-rs/src/flags.rs
+++ b/crates/storage/libmdbx-rs/src/flags.rs
@@ -199,7 +199,7 @@ impl EnvironmentFlags {
            flags |= ffi::MDBX_LIFORECLAIM;
        }

-        flags |= ffi::MDBX_NOTLS;
+        flags |= ffi::MDBX_NOSTICKYTHREADS;

        flags
    }
--- a/crates/storage/libmdbx-rs/src/txn_pool.rs
+++ b/crates/storage/libmdbx-rs/src/txn_pool.rs
@@ -3,8 +3,8 @@ use crossbeam_queue::ArrayQueue;

 /// Lock-free pool of reset read-only MDBX transaction handles.
 ///
-/// With `MDBX_NOTLS` (which reth always sets), every `mdbx_txn_begin_ex` for a read transaction
-/// calls `mvcc_bind_slot`, which acquires `lck_rdt_lock` — a pthread mutex. Under high
+/// With `MDBX_NOSTICKYTHREADS` (which reth always sets), every `mdbx_txn_begin_ex` for a read
+/// transaction calls `mvcc_bind_slot`, which acquires `lck_rdt_lock` — a pthread mutex. Under high
 /// concurrency (e.g., prewarming), this becomes a contention point.
 ///
 /// This pool caches transaction handles that have been reset via `mdbx_txn_reset`. A reset handle
--- a/crates/transaction-pool/src/blobstore/disk.rs
+++ b/crates/transaction-pool/src/blobstore/disk.rs
@@ -2,12 +2,12 @@

 use crate::blobstore::{BlobStore, BlobStoreCleanupStat, BlobStoreError, BlobStoreSize};
 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
-    eip7594::BlobTransactionSidecarVariant,
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
+    eip7594::{BlobCellMask, BlobTransactionSidecarVariant},
    eip7840::BlobParams,
    merge::EPOCH_SLOTS,
 };
-use alloy_primitives::{map::B256Set, TxHash, B256};
+use alloy_primitives::{map::B256Set, TxHash, B128, B256};
 use parking_lot::{Mutex, RwLock};
 use schnellru::{ByLength, LruMap};
 use std::{fmt, fs, io, path::PathBuf, sync::Arc};
@@ -133,6 +133,74 @@ impl DiskFileBlobStore {

        Ok(result)
    }
+
+    /// Look up EIP-7594 blob cells by their versioned hashes.
+    fn get_by_versioned_hashes_cells_eip7594(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        let cell_mask = BlobCellMask::new(indices_bitarray);
+        let mut result = vec![None; versioned_hashes.len()];
+        let mut missing_count = result.len();
+
+        let cached_blob_sidecars = self
+            .inner
+            .blob_cache
+            .lock()
+            .iter()
+            .map(|(_, blob_sidecar)| Arc::clone(blob_sidecar))
+            .collect::<Vec<_>>();
+        for blob_sidecar in cached_blob_sidecars {
+            if let Some(blob_sidecar) = blob_sidecar.as_eip7594() {
+                for (hash_idx, match_result) in blob_sidecar
+                    .match_versioned_hashes_cells(versioned_hashes, cell_mask)
+                    .map_err(|err| BlobStoreError::Other(Box::new(err)))?
+                {
+                    let slot = &mut result[hash_idx];
+                    if slot.is_none() {
+                        missing_count -= 1;
+                    }
+                    *slot = Some(match_result);
+                }
+            }
+
+            if missing_count == 0 && result.iter().all(Option::is_some) {
+                return Ok(result)
+            }
+        }
+
+        let mut missing_tx_hashes = Vec::new();
+        {
+            let mut versioned_to_txhashes = self.inner.versioned_hashes_to_txhash.lock();
+            for (idx, _) in
+                result.iter().enumerate().filter(|(_, cells_and_proofs)| cells_and_proofs.is_none())
+            {
+                let versioned_hash = versioned_hashes[idx];
+                if let Some(tx_hash) = versioned_to_txhashes.get(&versioned_hash).copied() {
+                    missing_tx_hashes.push(tx_hash);
+                }
+            }
+        }
+
+        if !missing_tx_hashes.is_empty() {
+            let blobs_from_disk = self.inner.read_many_decoded(missing_tx_hashes);
+            for (_, blob_sidecar) in blobs_from_disk {
+                if let Some(blob_sidecar) = blob_sidecar.as_eip7594() {
+                    for (hash_idx, match_result) in blob_sidecar
+                        .match_versioned_hashes_cells(versioned_hashes, cell_mask)
+                        .map_err(|err| BlobStoreError::Other(Box::new(err)))?
+                    {
+                        if result[hash_idx].is_none() {
+                            result[hash_idx] = Some(match_result);
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(result)
+    }
 }

 impl BlobStore for DiskFileBlobStore {
@@ -303,6 +371,14 @@ impl BlobStore for DiskFileBlobStore {
        self.get_by_versioned_hashes_eip7594(versioned_hashes)
    }

+    fn get_by_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        self.get_by_versioned_hashes_cells_eip7594(versioned_hashes, indices_bitarray)
+    }
+
    fn data_size_hint(&self) -> Option<usize> {
        Some(self.inner.size_tracker.data_size())
    }
@@ -925,6 +1001,27 @@ mod tests {
        assert_eq!(v3, vec![Some(expected), None]);
    }

+    #[test]
+    fn disk_get_blobs_v4_returns_requested_cells() {
+        let (store, _dir) = tmp_store();
+
+        let (sidecar, versioned_hash, _) = eip7594_single_blob_sidecar();
+        store.insert(TxHash::random(), sidecar).unwrap();
+
+        let indices_bitarray = B128::from((1u128 << 0) | (1u128 << 7));
+        let request = vec![versioned_hash, B256::ZERO];
+
+        let v4 = store.get_by_versioned_hashes_v4(&request, indices_bitarray).unwrap();
+        assert_eq!(v4.len(), request.len());
+        assert!(v4[1].is_none());
+
+        let cells_and_proofs = v4[0].as_ref().unwrap();
+        assert_eq!(cells_and_proofs.blob_cells.len(), 2);
+        assert_eq!(cells_and_proofs.proofs.len(), 2);
+        assert!(cells_and_proofs.blob_cells.iter().all(Option::is_some));
+        assert_eq!(cells_and_proofs.proofs, vec![Some(Bytes48::default()); 2]);
+    }
+
    #[test]
    fn disk_get_blobs_v3_can_fallback_to_disk() {
        let (store, _dir) = tmp_store();
@@ -937,6 +1034,20 @@ mod tests {
        assert_eq!(v3, vec![Some(expected)]);
    }

+    #[test]
+    fn disk_get_blobs_v4_can_fallback_to_disk() {
+        let (store, _dir) = tmp_store();
+
+        let (sidecar, versioned_hash, _) = eip7594_single_blob_sidecar();
+        store.insert(TxHash::random(), sidecar).unwrap();
+        store.clear_cache();
+
+        let v4 = store.get_by_versioned_hashes_v4(&[versioned_hash], B128::from(1u128)).unwrap();
+        let cells_and_proofs = v4[0].as_ref().unwrap();
+        assert_eq!(cells_and_proofs.blob_cells.len(), 1);
+        assert_eq!(cells_and_proofs.proofs, vec![Some(Bytes48::default())]);
+    }
+
    #[test]
    fn disk_double_cleanup_no_failure() {
        let (store, _dir) = tmp_store();
--- a/crates/transaction-pool/src/blobstore/mem.rs
+++ b/crates/transaction-pool/src/blobstore/mem.rs
@@ -1,9 +1,9 @@
 use crate::blobstore::{BlobStore, BlobStoreCleanupStat, BlobStoreError, BlobStoreSize};
 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
-    eip7594::BlobTransactionSidecarVariant,
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
+    eip7594::{BlobCellMask, BlobTransactionSidecarVariant},
 };
-use alloy_primitives::{map::B256Map, B256};
+use alloy_primitives::{map::B256Map, B128, B256};
 use parking_lot::RwLock;
 use std::sync::Arc;

@@ -48,6 +48,37 @@ impl InMemoryBlobStore {
        }
        result
    }
+
+    /// Look up EIP-7594 blob cells by their versioned hashes.
+    fn get_by_versioned_hashes_cells_eip7594(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        let cell_mask = BlobCellMask::new(indices_bitarray);
+        let mut result = vec![None; versioned_hashes.len()];
+        let mut missing_count = result.len();
+        let blob_sidecars = self.inner.store.read().values().cloned().collect::<Vec<_>>();
+        for blob_sidecar in blob_sidecars {
+            if let Some(blob_sidecar) = blob_sidecar.as_eip7594() {
+                for (hash_idx, match_result) in blob_sidecar
+                    .match_versioned_hashes_cells(versioned_hashes, cell_mask)
+                    .map_err(|err| BlobStoreError::Other(Box::new(err)))?
+                {
+                    let slot = &mut result[hash_idx];
+                    if slot.is_none() {
+                        missing_count -= 1;
+                    }
+                    *slot = Some(match_result);
+                }
+            }
+
+            if missing_count == 0 && result.iter().all(Option::is_some) {
+                break;
+            }
+        }
+        Ok(result)
+    }
 }

 #[derive(Debug, Default)]
@@ -186,6 +217,14 @@ impl BlobStore for InMemoryBlobStore {
        Ok(self.get_by_versioned_hashes_eip7594(versioned_hashes))
    }

+    fn get_by_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        self.get_by_versioned_hashes_cells_eip7594(versioned_hashes, indices_bitarray)
+    }
+
    fn data_size_hint(&self) -> Option<usize> {
        Some(self.inner.size_tracker.data_size())
    }
@@ -255,4 +294,25 @@ mod tests {
        let v3 = store.get_by_versioned_hashes_v3(&request).unwrap();
        assert_eq!(v3, vec![Some(expected), None]);
    }
+
+    #[test]
+    fn mem_get_blobs_v4_returns_requested_cells() {
+        let store = InMemoryBlobStore::default();
+
+        let (sidecar, versioned_hash, _) = eip7594_single_blob_sidecar();
+        store.insert(B256::random(), sidecar).unwrap();
+
+        let indices_bitarray = B128::from((1u128 << 0) | (1u128 << 7));
+        let request = vec![versioned_hash, B256::ZERO];
+
+        let v4 = store.get_by_versioned_hashes_v4(&request, indices_bitarray).unwrap();
+        assert_eq!(v4.len(), request.len());
+        assert!(v4[1].is_none());
+
+        let cells_and_proofs = v4[0].as_ref().unwrap();
+        assert_eq!(cells_and_proofs.blob_cells.len(), 2);
+        assert_eq!(cells_and_proofs.proofs.len(), 2);
+        assert!(cells_and_proofs.blob_cells.iter().all(Option::is_some));
+        assert_eq!(cells_and_proofs.proofs, vec![Some(Bytes48::default()); 2]);
+    }
 }
--- a/crates/transaction-pool/src/blobstore/mod.rs
+++ b/crates/transaction-pool/src/blobstore/mod.rs
@@ -1,10 +1,10 @@
 //! Storage for blob data of EIP4844 transactions.

 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip7594::BlobTransactionSidecarVariant,
 };
-use alloy_primitives::B256;
+use alloy_primitives::{B128, B256};
 pub use converter::BlobSidecarConverter;
 pub use disk::{DiskFileBlobStore, DiskFileBlobStoreConfig, OpenDiskFileBlobStore};
 pub use mem::InMemoryBlobStore;
@@ -109,6 +109,17 @@ pub trait BlobStore: fmt::Debug + Send + Sync + 'static {
        versioned_hashes: &[B256],
    ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError>;

+    /// Return the [`BlobCellsAndProofsV1`]s for a list of blob versioned hashes and requested cell
+    /// indices.
+    ///
+    /// The response is always the same length as the request. Missing or older-version blobs are
+    /// returned as `None` elements.
+    fn get_by_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError>;
+
    /// Data size of all transactions in the blob store.
    fn data_size_hint(&self) -> Option<usize>;

--- a/crates/transaction-pool/src/blobstore/noop.rs
+++ b/crates/transaction-pool/src/blobstore/noop.rs
@@ -1,9 +1,9 @@
 use crate::blobstore::{BlobStore, BlobStoreCleanupStat, BlobStoreError};
 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip7594::BlobTransactionSidecarVariant,
 };
-use alloy_primitives::B256;
+use alloy_primitives::{B128, B256};
 use std::sync::Arc;

 /// A blobstore implementation that does nothing
@@ -85,6 +85,14 @@ impl BlobStore for NoopBlobStore {
        Ok(vec![None; versioned_hashes.len()])
    }

+    fn get_by_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        _indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        Ok(vec![None; versioned_hashes.len()])
+    }
+
    fn data_size_hint(&self) -> Option<usize> {
        Some(0)
    }
--- a/crates/transaction-pool/src/lib.rs
+++ b/crates/transaction-pool/src/lib.rs
@@ -303,10 +303,10 @@ pub use crate::{
 };
 use crate::{identifier::TransactionId, pool::PoolInner};
 use alloy_eips::{
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip7594::BlobTransactionSidecarVariant,
 };
-use alloy_primitives::{map::AddressSet, Address, TxHash, B256, U256};
+use alloy_primitives::{map::AddressSet, Address, TxHash, B128, B256, U256};
 use aquamarine as _;
 use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
 use reth_eth_wire_types::HandleMempoolData;
@@ -808,6 +808,14 @@ where
    ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError> {
        self.pool.blob_store().get_by_versioned_hashes_v3(versioned_hashes)
    }
+
+    fn get_blobs_for_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        self.pool.blob_store().get_by_versioned_hashes_v4(versioned_hashes, indices_bitarray)
+    }
 }

 impl<V, T, S> TransactionPoolExt for Pool<V, T, S>
--- a/crates/transaction-pool/src/noop.rs
+++ b/crates/transaction-pool/src/noop.rs
@@ -16,10 +16,10 @@ use crate::{
 };
 use alloy_eips::{
    eip1559::ETHEREUM_BLOCK_GAS_LIMIT_30M,
-    eip4844::{BlobAndProofV1, BlobAndProofV2},
+    eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
    eip7594::BlobTransactionSidecarVariant,
 };
-use alloy_primitives::{map::AddressSet, Address, TxHash, B256, U256};
+use alloy_primitives::{map::AddressSet, Address, TxHash, B128, B256, U256};
 use reth_eth_wire_types::HandleMempoolData;
 use reth_primitives_traits::Recovered;
 use std::{marker::PhantomData, sync::Arc};
@@ -380,6 +380,14 @@ impl<T: EthPoolTransaction> TransactionPool for NoopTransactionPool<T> {
    ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError> {
        Ok(vec![None; versioned_hashes.len()])
    }
+
+    fn get_blobs_for_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        _indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
+        Ok(vec![None; versioned_hashes.len()])
+    }
 }

 /// A [`TransactionValidator`] that does nothing.
--- a/crates/transaction-pool/src/traits.rs
+++ b/crates/transaction-pool/src/traits.rs
@@ -65,12 +65,13 @@ use alloy_eips::{
    eip2718::{Encodable2718, WithEncoded},
    eip2930::AccessList,
    eip4844::{
-        env_settings::KzgSettings, BlobAndProofV1, BlobAndProofV2, BlobTransactionValidationError,
+        env_settings::KzgSettings, BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1,
+        BlobTransactionValidationError,
    },
    eip7594::BlobTransactionSidecarVariant,
    eip7702::SignedAuthorization,
 };
-use alloy_primitives::{map::AddressSet, Address, Bytes, TxHash, TxKind, B256, U256};
+use alloy_primitives::{map::AddressSet, Address, Bytes, TxHash, TxKind, B128, B256, U256};
 use futures_util::{ready, Stream};
 use reth_eth_wire_types::HandleMempoolData;
 use reth_ethereum_primitives::{PooledTransactionVariant, TransactionSigned};
@@ -722,6 +723,17 @@ pub trait TransactionPool: Clone + Debug + Send + Sync {
        &self,
        versioned_hashes: &[B256],
    ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError>;
+
+    /// Return the [`BlobCellsAndProofsV1`]s for a list of blob versioned hashes and requested cell
+    /// indices.
+    ///
+    /// The response is always the same length as the request. Missing or older-version blobs are
+    /// returned as `None` elements.
+    fn get_blobs_for_versioned_hashes_v4(
+        &self,
+        versioned_hashes: &[B256],
+        indices_bitarray: B128,
+    ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError>;
 }

 /// Extension for [`TransactionPool`] trait that allows to set the current block info.
--- a/crates/transaction-pool/src/validate/eth.rs
+++ b/crates/transaction-pool/src/validate/eth.rs
@@ -1424,7 +1424,6 @@ pub fn ensure_intrinsic_gas<T: EthPoolTransaction>(
            .map(|l| l.iter().map(|i| i.storage_keys.len()).sum::<usize>())
            .unwrap_or_default() as u64,
        transaction.authorization_list().map(|l| l.len()).unwrap_or_default() as u64,
-        revm_primitives::eip8037::CPSB_GLAMSTERDAM,
    );

    let gas_limit = transaction.gas_limit();
--- a/docs/vocs/docs/pages/cli/reth/node.mdx
+++ b/docs/vocs/docs/pages/cli/reth/node.mdx
@@ -115,7 +115,9 @@ Networking:
          [default: 9200]

      --discovery.v5.port.ipv6 <DISCOVERY_V5_PORT_IPV6>
-          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set
+          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set.
+
+          If not provided, discovery V5 defaults to same port as discovery V4 (--discovery.port).

          [default: 9200]

@@ -1090,7 +1092,7 @@ Engine:
          Disable BAL-driven parallel state root computation. When set, the BAL hashed post state is not sent to the multiproof task for early parallel state root computation

      --engine.disable-bal-batch-io
-          Disable BAL (Block Access List) batched IO during prewarming. When set, falls back to individual per-slot storage reads instead of batched cursor reads
+          Disable BAL (Block Access List) storage prefetch IO during prewarming. When set, BAL storage slots are not read into the execution cache

 ERA:
      --era.enable
--- a/docs/vocs/docs/pages/cli/reth/p2p/body.mdx
+++ b/docs/vocs/docs/pages/cli/reth/p2p/body.mdx
@@ -55,7 +55,9 @@ Networking:
          [default: 9200]

      --discovery.v5.port.ipv6 <DISCOVERY_V5_PORT_IPV6>
-          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set
+          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set.
+
+          If not provided, discovery V5 defaults to same port as discovery V4 (--discovery.port).

          [default: 9200]

--- a/docs/vocs/docs/pages/cli/reth/p2p/header.mdx
+++ b/docs/vocs/docs/pages/cli/reth/p2p/header.mdx
@@ -55,7 +55,9 @@ Networking:
          [default: 9200]

      --discovery.v5.port.ipv6 <DISCOVERY_V5_PORT_IPV6>
-          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set
+          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set.
+
+          If not provided, discovery V5 defaults to same port as discovery V4 (--discovery.port).

          [default: 9200]

--- a/docs/vocs/docs/pages/cli/reth/stage/run.mdx
+++ b/docs/vocs/docs/pages/cli/reth/stage/run.mdx
@@ -208,7 +208,9 @@ Networking:
          [default: 9200]

      --discovery.v5.port.ipv6 <DISCOVERY_V5_PORT_IPV6>
-          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set
+          The UDP IPv6 port to use for devp2p peer discovery version 5. Not used unless `--addr` is IPv6, or `--discovery.addr.ipv6` is set.
+
+          If not provided, discovery V5 defaults to same port as discovery V4 (--discovery.port).

          [default: 9200]

--- a/docs/vocs/vocs.config.ts
+++ b/docs/vocs/vocs.config.ts
@@ -22,7 +22,7 @@ export default defineConfig({
    },
    { text: 'GitHub', link: 'https://github.com/paradigmxyz/reth' },
    {
-      text: 'v2.1.0',
+      text: 'v2.2.0',
      items: [
        {
          text: 'Releases',
--- a/testing/ef-tests/src/cases/blockchain_test.rs
+++ b/testing/ef-tests/src/cases/blockchain_test.rs
@@ -252,7 +252,7 @@ fn run_case(case: &BlockchainTest) -> Result<(), Error> {
            .map_err(|err| Error::block_failed(block_number, err))?;

        // Consensus checks after block execution
-        validate_block_post_execution(block, &chain_spec, &output, None, None)
+        validate_block_post_execution(block, &chain_spec, &output, None)
            .map_err(|err| Error::block_failed(block_number, err))?;

        // Compute and check the post state root
Author	SHA1	Message	Date
DaniPopes	fcfa8287f6	chore(mdbx): replace deprecated MDBX_NOTLS with MDBX_NOSTICKYTHREADS (#23378 )	2026-04-30 03:19:09 +00:00
Arsenii Kulikov	d25de30050	feat: customizable discovery defaults (#23843 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2026-04-29 22:59:10 +00:00
Derek Cofausper	4ffde69d94	fix(engine): apply finalized state after syncing FCU head import (#23838 ) Co-authored-by: Centaur AI <ai@centaur.local> Co-authored-by: Amp <amp@ampcode.com> Co-authored-by: klkvr <klkvrr@gmail.com>	2026-04-29 15:29:42 +00:00
Arsenii Kulikov	077e5eecfe	chore: don't enforce non-empty blocks in e2e payload building (#23837 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2026-04-29 14:36:44 +00:00
Sergei Shulepov	709485dcb7	perf(bench): buffer RPC fetches in generate-big-block (#23830 ) Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>	2026-04-29 12:57:05 +00:00
Arsenii Kulikov	88505c7fcb	fix(re-execute): properly handle selfdestructed storage slots (#23832 )	2026-04-29 12:30:27 +00:00
Emma Jamieson-Hoare	c14bc59236	chore: release 2.2.0 (#23831 ) Co-authored-by: Amp <amp@ampcode.com>	2026-04-29 12:25:15 +00:00
Derek Cofausper	347c1325cc	fix: skip `move_to_static_files` for `storage.v2` (#23814 )	2026-04-29 12:19:43 +00:00
Matthias Seitz	5f85eb7ac8	feat(engine): add getBlobsV4 endpoint (#23767 )	2026-04-29 12:02:25 +00:00
Brian Picciano	a12454d2e6	perf(db): prebind cursor operation metrics (#23654 ) Co-authored-by: Amp <amp@ampcode.com> Co-authored-by: Brian Picciano <933154+mediocregopher@users.noreply.github.com>	2026-04-29 11:40:17 +00:00
Matthias Seitz	c194c17a27	chore(deps): bump alloy to 2.0.4 (#23828 )	2026-04-29 13:41:02 +02:00
figtracer	43a7452b0e	fix(rpc): narrow getLogs retry range (#23818 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2026-04-29 10:09:49 +00:00
Matthias Seitz	73ec2c9d56	docs(engine): clarify BAL storage prefetch flag (#23815 )	2026-04-29 08:37:51 +00:00
Alexey Shekhirin	76e886578b	ci(hive): update amsterdam fixtures and branch (#23807 )	2026-04-29 07:48:24 +00:00