error -> debug

This reverts commit 8c6080bb2b.

.github/workflows/integration.yml

@@ -22,7 +22,7 @@ concurrency:
 
 jobs:
   test:
-    name: test / ${{ matrix.network }} / ${{ matrix.storage }}
+    name: test / ${{ matrix.network }}
     if: github.event_name != 'schedule'
     runs-on: depot-ubuntu-latest-4
     env:
@@ -30,10 +30,6 @@ jobs:
     strategy:
       matrix:
         network: ["ethereum", "optimism"]
-        storage: ["stable", "edge"]
-        exclude:
-          - network: optimism
-            storage: edge
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v6
@@ -50,7 +46,7 @@ jobs:
         name: Run tests
         run: |
           cargo nextest run \
-            --locked --features "asm-keccak ${{ matrix.network }} ${{ matrix.storage == 'edge' && 'edge' || '' }}" \
+            --locked --features "asm-keccak ${{ matrix.network }}" \
             --workspace --exclude ef-tests \
             -E "kind(test) and not binary(e2e_testsuite)"
       - if: matrix.network == 'optimism'

CLAUDE.md

@@ -38,7 +38,7 @@ Reth is a high-performance Ethereum execution client written in Rust, focusing o
 
 2. **Linting**: Run clippy with all features
    ```bash
-   cargo +nightly clippy --workspace --lib --examples --tests --benches --all-features 
+   RUSTFLAGS="-D warnings" cargo +nightly clippy --workspace --lib --examples --tests --benches --all-features --locked
    ```
 
 3. **Testing**: Use nextest for faster test execution
@@ -169,11 +169,12 @@ Based on PR patterns, avoid:
 Before submitting changes, ensure:
 
 1. **Format Check**: `cargo +nightly fmt --all --check`
-2. **Clippy**: No warnings
+2. **Clippy**: No warnings with `RUSTFLAGS="-D warnings"`
 3. **Tests Pass**: All unit and integration tests
 4. **Documentation**: Update relevant docs and add doc comments with `cargo docs --document-private-items`
 5. **Commit Messages**: Follow conventional format (feat:, fix:, chore:, etc.)
 
+
 ### Opening PRs against <https://github.com/paradigmxyz/reth>
 
 Label PRs appropriately, first check the available labels and then apply the relevant ones:
@@ -348,10 +349,10 @@ Let's say you want to fix a bug where external IP resolution fails on startup:
    }
    ```
 
-5. **Run checks** (IMPORTANT!):
+5. **Run checks**:
    ```bash
    cargo +nightly fmt --all
-   cargo clippy --workspace --all-features # Make sure WHOLE WORKSPACE compiles!
+   cargo clippy --all-features
    cargo test -p reth-discv4
    ```
 
@@ -373,7 +374,7 @@ Let's say you want to fix a bug where external IP resolution fails on startup:
 cargo +nightly fmt --all
 
 # Run lints
-cargo +nightly clippy --workspace --all-features
+RUSTFLAGS="-D warnings" cargo +nightly clippy --workspace --all-features --locked
 
 # Run tests
 cargo nextest run --workspace

Cargo.toml

@@ -77,10 +77,6 @@ members = [
     "crates/optimism/cli",
     "crates/optimism/consensus",
     "crates/optimism/evm/",
-    "crates/optimism/examples/custom-node",
-    "crates/optimism/examples/engine-api-access",
-    "crates/optimism/examples/exex-hello-world",
-    "crates/optimism/examples/op-db-access",
     "crates/optimism/flashblocks/",
     "crates/optimism/hardforks/",
     "crates/optimism/node/",
@@ -149,6 +145,7 @@ members = [
     "examples/beacon-api-sse/",
     "examples/bsc-p2p",
     "examples/custom-dev-node/",
+    "examples/custom-node/",
     "examples/custom-engine-types/",
     "examples/custom-evm/",
     "examples/custom-hardforks/",
@@ -157,7 +154,10 @@ members = [
     "examples/custom-payload-builder/",
     "examples/custom-rlpx-subprotocol",
     "examples/custom-rpc-middleware",
+    "examples/custom-node",
     "examples/db-access",
+    "examples/engine-api-access",
+    "examples/exex-hello-world",
     "examples/exex-subscription",
     "examples/exex-test",
     "examples/full-contract-state",
@@ -168,6 +168,7 @@ members = [
     "examples/node-builder-api/",
     "examples/node-custom-rpc/",
     "examples/node-event-hooks/",
+    "examples/op-db-access/",
     "examples/polygon-p2p/",
     "examples/rpc-db/",
     "examples/precompile-cache/",
@@ -560,7 +561,7 @@ humantime-serde = "1.1"
 itertools = { version = "0.14", default-features = false }
 linked_hash_set = "0.1"
 lz4 = "1.28.1"
-modular-bitfield = "0.13.1"
+modular-bitfield = "0.11.2"
 notify = { version = "8.0.0", default-features = false, features = ["macos_fsevent"] }
 nybbles = { version = "0.4.2", default-features = false }
 once_cell = { version = "1.19", default-features = false, features = ["critical-section"] }
@@ -589,13 +590,13 @@ zstd = "0.13"
 byteorder = "1"
 fixed-cache = { version = "0.1.7", features = ["stats"] }
 moka = "0.12"
-tar-no-std = { version = "0.4.2", default-features = false }
-miniz_oxide = { version = "0.9.0", default-features = false }
+tar-no-std = { version = "0.3.2", default-features = false }
+miniz_oxide = { version = "0.8.4", default-features = false }
 chrono = "0.4.41"
 
 # metrics
 metrics = "0.24.0"
-metrics-derive = "0.1.1"
+metrics-derive = "0.1"
 metrics-exporter-prometheus = { version = "0.18.0", default-features = false }
 metrics-process = "2.1.0"
 metrics-util = { default-features = false, version = "0.20.0" }
@@ -607,7 +608,7 @@ quote = "1.0"
 # tokio
 tokio = { version = "1.44.2", default-features = false }
 tokio-stream = "0.1.11"
-tokio-tungstenite = "0.28.0"
+tokio-tungstenite = "0.26.2"
 tokio-util = { version = "0.7.4", features = ["codec"] }
 
 # async
@@ -620,7 +621,7 @@ futures-util = { version = "0.3", default-features = false }
 hyper = "1.3"
 hyper-util = "0.1.5"
 pin-project = "1.0.12"
-reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "rustls-tls-native-roots", "stream"] }
+reqwest = { version = "0.12", default-features = false }
 tracing-futures = "0.2"
 tower = "0.5"
 tower-http = "0.6"
@@ -640,6 +641,7 @@ jsonrpsee-types = "0.26.0"
 http = "1.0"
 http-body = "1.0"
 http-body-util = "0.1.2"
+jsonwebtoken = "9"
 proptest-arbitrary-interop = "0.1.0"
 
 # crypto
@@ -653,7 +655,7 @@ rand_08 = { package = "rand", version = "0.8" }
 c-kzg = "2.1.5"
 
 # config
-toml = "0.9"
+toml = "0.8"
 
 # rocksdb
 rocksdb = { version = "0.24" }
@@ -669,19 +671,19 @@ tracing-opentelemetry = "0.32"
 # misc-testing
 arbitrary = "1.3"
 assert_matches = "1.5.0"
-criterion = { package = "codspeed-criterion-compat", version = "4.3" }
+criterion = { package = "codspeed-criterion-compat", version = "2.7" }
 insta = "1.41"
 proptest = "1.7"
-proptest-derive = "0.7"
+proptest-derive = "0.5"
 similar-asserts = { version = "1.5.0", features = ["serde"] }
 tempfile = "3.20"
 test-fuzz = "7"
-rstest = "0.26.1"
+rstest = "0.24.0"
 test-case = "3"
 
 # ssz encoding
-ethereum_ssz = "0.10.1"
-ethereum_ssz_derive = "0.10.1"
+ethereum_ssz = "0.9.0"
+ethereum_ssz_derive = "0.9.0"
 
 # allocators
 jemalloc_pprof = { version = "0.8", default-features = false }
@@ -693,14 +695,14 @@ snmalloc-rs = { version = "0.3.7", features = ["build_cc"] }
 aes = "0.8.1"
 ahash = "0.8"
 anyhow = "1.0"
-bindgen = { version = "0.72", default-features = false }
-block-padding = "0.3"
+bindgen = { version = "0.71", default-features = false }
+block-padding = "0.3.2"
 cc = "1.2.15"
 cipher = "0.4.3"
 comfy-table = "7.0"
 concat-kdf = "0.1.0"
 crossbeam-channel = "0.5.13"
-crossterm = "0.29.0"
+crossterm = "0.28.0"
 csv = "1.3.0"
 ctrlc = "3.4"
 ctr = "0.9.2"
@@ -713,7 +715,7 @@ hmac = "0.12.1"
 human_bytes = "0.4.1"
 indexmap = "2"
 interprocess = "2.2.0"
-lz4_flex = { version = "0.12", default-features = false }
+lz4_flex = { version = "0.11", default-features = false }
 memmap2 = "0.9.4"
 mev-share-sse = { version = "0.5.0", default-features = false }
 num-traits = "0.2.15"
@@ -721,15 +723,15 @@ page_size = "0.6.0"
 parity-scale-codec = "3.2.1"
 plain_hasher = "0.2"
 pretty_assertions = "1.4"
-ratatui = { version = "0.30", default-features = false }
-ringbuffer = "0.16.0"
+ratatui = { version = "0.29", default-features = false }
+ringbuffer = "0.15.0"
 rmp-serde = "1.3"
-roaring = "0.11.3"
+roaring = "0.10.2"
 rolling-file = "0.2.0"
 sha3 = "0.10.5"
 snap = "1.1.1"
-socket2 = { version = "0.6", default-features = false }
-sysinfo = { version = "0.38", default-features = false }
+socket2 = { version = "0.5", default-features = false }
+sysinfo = { version = "0.33", default-features = false }
 tracing-journald = "0.3"
 tracing-logfmt = "=0.3.5"
 tracing-samply = "0.1"

Dockerfile.depot

@@ -5,7 +5,7 @@
 #   reth:    --build-arg BINARY=reth
 #   op-reth: --build-arg BINARY=op-reth --build-arg MANIFEST_PATH=crates/optimism/bin
 
-FROM rust:1 AS builder
+FROM lukemathwalker/cargo-chef:latest-rust-1 AS chef
 WORKDIR /app
 
 LABEL org.opencontainers.image.source=https://github.com/paradigmxyz/reth
@@ -19,6 +19,14 @@ ENV RUSTC_WRAPPER=sccache
 ENV SCCACHE_DIR=/sccache
 ENV SCCACHE_WEBDAV_ENDPOINT=https://cache.depot.dev
 
+# Builds a cargo-chef plan
+FROM chef AS planner
+COPY --exclude=.git . .
+RUN cargo chef prepare --recipe-path recipe.json
+
+FROM chef AS builder
+COPY --from=planner /app/recipe.json recipe.json
+
 # Binary to build (reth or op-reth)
 ARG BINARY=reth
 
@@ -45,6 +53,13 @@ ENV VERGEN_GIT_SHA=$VERGEN_GIT_SHA
 ENV VERGEN_GIT_DESCRIBE=$VERGEN_GIT_DESCRIBE
 ENV VERGEN_GIT_DIRTY=$VERGEN_GIT_DIRTY
 
+# Build dependencies
+RUN --mount=type=secret,id=DEPOT_TOKEN,env=SCCACHE_WEBDAV_TOKEN \
+    --mount=type=cache,target=/usr/local/cargo/registry,sharing=shared \
+    --mount=type=cache,target=/usr/local/cargo/git,sharing=shared \
+    --mount=type=cache,target=$SCCACHE_DIR,sharing=shared \
+    cargo chef cook --profile $BUILD_PROFILE --features "$FEATURES" --locked --recipe-path recipe.json --manifest-path $MANIFEST_PATH/Cargo.toml
+
 # Build application
 COPY --exclude=.git . .
 RUN --mount=type=secret,id=DEPOT_TOKEN,env=SCCACHE_WEBDAV_TOKEN \

bin/reth-bench-compare/Cargo.toml

@@ -56,7 +56,7 @@ ctrlc.workspace = true
 shlex.workspace = true
 
 [target.'cfg(unix)'.dependencies]
-nix = { version = "0.31", features = ["signal", "process"] }
+nix = { version = "0.29", features = ["signal", "process"] }
 
 [features]
 default = ["jemalloc"]

bin/reth-bench-compare/src/cli.rs

@@ -274,10 +274,10 @@ impl Args {
     /// Get the default RPC URL for a given chain
     const fn get_default_rpc_url(chain: &Chain) -> &'static str {
         match chain.id() {
-            8453 => "https://base.reth.rs/rpc",             // base
+            8453 => "https://base-mainnet.rpc.ithaca.xyz",  // base
             84532 => "https://base-sepolia.rpc.ithaca.xyz", // base-sepolia
             27082 => "https://rpc.hoodi.ethpandaops.io",    // hoodi
-            _ => "https://ethereum.reth.rs/rpc",            // mainnet and fallback
+            _ => "https://reth-ethereum.ithaca.xyz/rpc",    // mainnet and fallback
         }
     }
 

bin/reth-bench/Cargo.toml

@@ -45,7 +45,7 @@ op-alloy-consensus = { workspace = true, features = ["alloy-compat"] }
 op-alloy-rpc-types-engine = { workspace = true, features = ["serde"] }
 
 # reqwest
-reqwest.workspace = true
+reqwest = { workspace = true, default-features = false, features = ["rustls-tls-native-roots"] }
 
 # tower
 tower.workspace = true

bin/reth-bench/README.md

@@ -47,60 +47,10 @@ Make sure `reth` is running in the background with the proper configuration. Thi
 
 Any consensus layer client configured to connect to `reth` should be shut down, as all the engine API interactions during benchmarking will be driven by `reth-bench`.
 
-#### Recommended: schelk Snapshot Workflow
-
-For fast, reproducible benchmarks with quick state reset between runs, use [`schelk`](https://github.com/tempoxyz/schelk) - a filesystem snapshot tool that enables ~10 second rollback times.
-
-**One-time setup** (requires two NVMe drives of equal size and a ramdisk):
-```bash
-# Load ramdisk module (4 GiB for ~1.7 TiB drives)
-sudo modprobe brd rd_size=4194304
-
-# Copy virgin state to scratch drive
-sudo dd if=/dev/nvme1n1 of=/dev/nvme2n1 bs=256M status=progress conv=fsync
-
-# Initialize schelk
-sudo schelk init \
-  --virgin /dev/nvme1n1 \
-  --scratch /dev/nvme2n1 \
-  --ramdisk /dev/ram0 \
-  --mount-point /data/reth \
-  --fstype ext4 \
-  --state-path ./schelk_state.json
-```
-
-**Benchmark workflow**:
-```bash
-# Mount the scratch volume
-sudo schelk mount
-
-# Start reth on the mounted datadir
-reth node --datadir /data/reth --authrpc.jwtsecret <jwt_file_path>
-
-# Run benchmark
-reth-bench new-payload-fcu --rpc-url <rpc-url> --from <start_block> --to <end_block> --jwt-secret <jwt_file_path>
-
-# Stop reth, then recover to pristine state (~10s)
-sudo schelk recover
-
-# Repeat for next benchmark run
-```
-
-This workflow is preferred because:
-- ~10 second rollback time vs minutes for `stage unwind`
-- Guaranteed identical starting state between runs
-- No database corruption risks from unwinding
-- Enables rapid A/B testing of reth changes
-
-#### Alternative: Stage Unwind Workflow
-
-> [!WARNING]
-> Use this workflow **only when specifically testing unwind/reorg codepaths**.
-> Running `reth stage unwind` on large databases with big blocks may trigger `UnsortedInput` errors in RocksDB history tables.
-> For performance benchmarking, prefer the schelk workflow above.
-
-If you need to test unwind/reorg behavior, or don't have the hardware for schelk, you can use `stage unwind`:
+Depending on the block range you want to use in the benchmark you may need to unwind your node.
+The head of the node should be behind the lowest block in the range.
 
+Starting with a synced ethereum mainnet node, to run a benchmark starting at block 21,000,000, the node would need to be unwound first:
 ```bash
 reth stage unwind to-block 21000000
 ```
@@ -110,9 +60,7 @@ The following `reth-bench` command would then start the benchmark at block 21,00
 reth-bench new-payload-fcu --rpc-url <rpc-url> --from 21000000 --to <end_block> --jwt-secret <jwt_file_path>
 ```
 
-#### Build Profiles
-
-Make sure that reth is built using a build profile suitable for what you are trying to measure.
+Finally, make sure that reth is built using a build profile suitable for what you are trying to measure.
 For example, if the purpose of the benchmark is to load test and measure `reth`'s maximum speed, it would be compiled with the `maxperf` profile:
 ```bash
 make maxperf
@@ -196,16 +144,12 @@ usage over time, and many other metrics that are useful for diagnosing performan
 
 ### Repeat
 
-To reproduce the benchmark, reset the node to the starting block:
-- **schelk workflow (recommended)**: Run `sudo schelk recover` (~10 seconds)
-- **Stage unwind workflow**: Use `reth stage unwind` to roll back to the starting block (see warnings above about potential issues)
-
-Then repeat all of the above steps.
+To reproduce the benchmark, first re-set the node to the block that the benchmark started at, using `reth stage unwind` as mentioned above, and repeat all of the above steps.
 
 ## Additional Considerations
 
 - **RPC Configuration**: The RPC endpoints should be accessible and configured correctly, specifically the RPC endpoint must support `eth_getBlockByNumber` and support fetching full transactions. The benchmark will make one RPC query per block as fast as possible, so ensure the RPC endpoint does not rate limit or block requests after a certain volume.
-- **Reproducibility**: Ensure that the node is at the same state before attempting to retry a benchmark. The `new-payload-fcu` command specifically will commit to the database, so the node must be reset. Use `schelk recover` (recommended, ~10s) or `reth stage unwind` (for reorg testing only) to reproducibly retry benchmarks.
+- **Reproducibility**: Ensure that the node is at the same state before attempting to retry a benchmark. The `new-payload-fcu` command specifically will commit to the database, so the node must be rolled back using `reth stage unwind` to reproducibly retry benchmarks.
 - **Profiling tools**: If you are collecting CPU profiles, tools like [`samply`](https://github.com/mstange/samply) and [`perf`](https://perf.wiki.kernel.org/index.php/Main_Page) can be useful for analyzing node performance.
 - **Benchmark Data**: `reth-bench` additionally contains a `--output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
 - **Platform Information**: To ensure accurate and reproducible benchmarking, document the platform details, including hardware specifications, OS version, and any other relevant information before publishing any benchmarks.

bin/reth-bench/src/bench/generate_big_block.rs

@@ -132,24 +132,13 @@ impl<S: TransactionSource> TransactionCollector<S> {
     /// Collect transactions starting from the given block number.
     ///
     /// Skips blob transactions (type 3) and collects until target gas is reached.
-    /// Returns a `CollectionResult` with transactions, gas info, and next block.
-    pub async fn collect(&self, start_block: u64) -> eyre::Result<CollectionResult> {
-        self.collect_gas(start_block, self.target_gas).await
-    }
-
-    /// Collect transactions up to a specific gas target.
-    ///
-    /// This is used both for initial collection and for retry top-ups.
-    pub async fn collect_gas(
-        &self,
-        start_block: u64,
-        gas_target: u64,
-    ) -> eyre::Result<CollectionResult> {
-        let mut transactions: Vec<RawTransaction> = Vec::new();
+    /// Returns the collected raw transaction bytes, total gas used, and the next block number.
+    pub async fn collect(&self, start_block: u64) -> eyre::Result<(Vec<Bytes>, u64, u64)> {
+        let mut transactions: Vec<Bytes> = Vec::new();
         let mut total_gas: u64 = 0;
         let mut current_block = start_block;
 
-        while total_gas < gas_target {
+        while total_gas < self.target_gas {
             let Some((block_txs, _)) = self.source.fetch_block_transactions(current_block).await?
             else {
                 warn!(block = current_block, "Block not found, stopping");
@@ -162,12 +151,12 @@ impl<S: TransactionSource> TransactionCollector<S> {
                     continue;
                 }
 
-                if total_gas + tx.gas_used <= gas_target {
+                if total_gas + tx.gas_used <= self.target_gas {
+                    transactions.push(tx.raw);
                     total_gas += tx.gas_used;
-                    transactions.push(tx);
                 }
 
-                if total_gas >= gas_target {
+                if total_gas >= self.target_gas {
                     break;
                 }
             }
@@ -175,7 +164,7 @@ impl<S: TransactionSource> TransactionCollector<S> {
             current_block += 1;
 
             // Stop early if remaining gas is under 1M (close enough to target)
-            let remaining_gas = gas_target.saturating_sub(total_gas);
+            let remaining_gas = self.target_gas.saturating_sub(total_gas);
             if remaining_gas < 1_000_000 {
                 break;
             }
@@ -183,12 +172,12 @@ impl<S: TransactionSource> TransactionCollector<S> {
 
         info!(
             total_txs = transactions.len(),
-            gas_sent = total_gas,
+            total_gas,
             next_block = current_block,
             "Finished collecting transactions"
         );
 
-        Ok(CollectionResult { transactions, gas_sent: total_gas, next_block: current_block })
+        Ok((transactions, total_gas, current_block))
     }
 }
 
@@ -263,80 +252,6 @@ struct BuiltPayload {
     envelope: ExecutionPayloadEnvelopeV4,
     block_hash: B256,
     timestamp: u64,
-    /// The actual gas used in the built block.
-    gas_used: u64,
-}
-
-/// Result of collecting transactions from blocks.
-#[derive(Debug)]
-pub struct CollectionResult {
-    /// Collected transactions with their gas info.
-    pub transactions: Vec<RawTransaction>,
-    /// Total gas sent (sum of historical `gas_used` for all collected txs).
-    pub gas_sent: u64,
-    /// Next block number to continue collecting from.
-    pub next_block: u64,
-}
-
-/// Constants for retry logic.
-const MAX_BUILD_RETRIES: u32 = 5;
-/// Maximum retries for fetching a transaction batch.
-const MAX_FETCH_RETRIES: u32 = 5;
-/// Tolerance: if `gas_used` is within 1M of target, don't retry.
-const MIN_TARGET_SLACK: u64 = 1_000_000;
-/// Maximum gas to request in retries (10x target as safety cap).
-const MAX_ADDITIONAL_GAS_MULTIPLIER: u64 = 10;
-
-/// Fetches a batch of transactions with retry logic.
-///
-/// Returns `None` if all retries are exhausted.
-async fn fetch_batch_with_retry<S: TransactionSource>(
-    collector: &TransactionCollector<S>,
-    block: u64,
-) -> Option<CollectionResult> {
-    for attempt in 1..=MAX_FETCH_RETRIES {
-        match collector.collect(block).await {
-            Ok(result) => return Some(result),
-            Err(e) => {
-                if attempt == MAX_FETCH_RETRIES {
-                    warn!(attempt, error = %e, "Failed to fetch transactions after max retries");
-                    return None;
-                }
-                warn!(attempt, error = %e, "Failed to fetch transactions, retrying...");
-                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-            }
-        }
-    }
-    None
-}
-
-/// Outcome of a build attempt check.
-enum RetryOutcome {
-    /// Payload is close enough to target gas.
-    Success,
-    /// Max retries reached, accept what we have.
-    MaxRetries,
-    /// Need more transactions with the specified gas amount.
-    NeedMore(u64),
-}
-
-/// Buffer for receiving transaction batches from the fetcher.
-///
-/// This abstracts over the channel to allow the main loop to request
-/// batches on demand, including for retries.
-struct TxBuffer {
-    receiver: mpsc::Receiver<CollectionResult>,
-}
-
-impl TxBuffer {
-    const fn new(receiver: mpsc::Receiver<CollectionResult>) -> Self {
-        Self { receiver }
-    }
-
-    /// Take the next available batch from the fetcher.
-    async fn take_batch(&mut self) -> Option<CollectionResult> {
-        self.receiver.recv().await
-    }
 }
 
 impl Command {
@@ -397,20 +312,19 @@ impl Command {
             )
             .await?;
         } else {
-            // Single payload - collect transactions and build with retry
+            // Single payload - collect transactions and build
             let tx_source = RpcTransactionSource::from_url(&self.rpc_url)?;
             let collector = TransactionCollector::new(tx_source, self.target_gas);
-            let result = collector.collect(start_block).await?;
+            let (transactions, _total_gas, _next_block) = collector.collect(start_block).await?;
 
-            if result.transactions.is_empty() {
+            if transactions.is_empty() {
                 return Err(eyre::eyre!("No transactions collected"));
             }
 
-            self.execute_sequential_with_retry(
+            self.execute_sequential(
                 &auth_provider,
                 &testing_provider,
-                &collector,
-                result,
+                transactions,
                 parent_hash,
                 parent_timestamp,
             )
@@ -421,34 +335,32 @@ impl Command {
         Ok(())
     }
 
-    /// Sequential execution path with retry logic for underfilled payloads.
-    async fn execute_sequential_with_retry<S: TransactionSource>(
+    /// Sequential execution path for single payload or no-execute mode.
+    async fn execute_sequential(
         &self,
         auth_provider: &RootProvider<AnyNetwork>,
         testing_provider: &RootProvider<AnyNetwork>,
-        collector: &TransactionCollector<S>,
-        initial_result: CollectionResult,
+        transactions: Vec<Bytes>,
         mut parent_hash: B256,
         mut parent_timestamp: u64,
     ) -> eyre::Result<()> {
-        let mut current_result = initial_result;
-
         for i in 0..self.count {
+            info!(
+                payload = i + 1,
+                total = self.count,
+                parent_hash = %parent_hash,
+                parent_timestamp = parent_timestamp,
+                "Building payload via testing_buildBlockV1"
+            );
+
             let built = self
-                .build_with_retry(
-                    testing_provider,
-                    collector,
-                    &mut current_result,
-                    i,
-                    parent_hash,
-                    parent_timestamp,
-                )
+                .build_payload(testing_provider, &transactions, i, parent_hash, parent_timestamp)
                 .await?;
 
             self.save_payload(&built)?;
 
             if self.execute || self.count > 1 {
-                info!(payload = i + 1, block_hash = %built.block_hash, gas_used = built.gas_used, "Executing payload (newPayload + FCU)");
+                info!(payload = i + 1, block_hash = %built.block_hash, "Executing payload (newPayload + FCU)");
                 self.execute_payload_v4(auth_provider, built.envelope, parent_hash).await?;
                 info!(payload = i + 1, "Payload executed successfully");
             }
@@ -459,62 +371,7 @@ impl Command {
         Ok(())
     }
 
-    /// Build a payload with retry logic when `gas_used` is below target.
-    ///
-    /// Uses the ratio of `gas_used/gas_sent` to estimate how many more transactions
-    /// are needed to hit the target gas.
-    async fn build_with_retry<S: TransactionSource>(
-        &self,
-        testing_provider: &RootProvider<AnyNetwork>,
-        collector: &TransactionCollector<S>,
-        result: &mut CollectionResult,
-        index: u64,
-        parent_hash: B256,
-        parent_timestamp: u64,
-    ) -> eyre::Result<BuiltPayload> {
-        for attempt in 1..=MAX_BUILD_RETRIES {
-            let tx_bytes: Vec<Bytes> = result.transactions.iter().map(|t| t.raw.clone()).collect();
-            let gas_sent = result.gas_sent;
-
-            info!(
-                payload = index + 1,
-                attempt,
-                tx_count = tx_bytes.len(),
-                gas_sent,
-                parent_hash = %parent_hash,
-                "Building payload via testing_buildBlockV1"
-            );
-
-            let built = Self::build_payload_static(
-                testing_provider,
-                &tx_bytes,
-                index,
-                parent_hash,
-                parent_timestamp,
-            )
-            .await?;
-
-            match self.check_retry_outcome(&built, index, attempt, gas_sent) {
-                RetryOutcome::Success | RetryOutcome::MaxRetries => return Ok(built),
-                RetryOutcome::NeedMore(additional_gas) => {
-                    let additional =
-                        collector.collect_gas(result.next_block, additional_gas).await?;
-                    result.transactions.extend(additional.transactions);
-                    result.gas_sent = result.gas_sent.saturating_add(additional.gas_sent);
-                    result.next_block = additional.next_block;
-                }
-            }
-        }
-
-        warn!(payload = index + 1, "Retry loop exited without returning a payload");
-        Err(eyre::eyre!("build_with_retry exhausted retries without result"))
-    }
-
-    /// Pipelined execution - fetches transactions in background, builds with retry.
-    ///
-    /// The fetcher continuously produces transaction batches. The main loop consumes them,
-    /// builds payloads with retry logic (requesting more transactions if underfilled),
-    /// and executes each payload before moving to the next.
+    /// Pipelined execution - fetches transactions and builds payloads in background.
     async fn execute_pipelined(
         &self,
         auth_provider: &RootProvider<AnyNetwork>,
@@ -523,220 +380,180 @@ impl Command {
         initial_parent_hash: B256,
         initial_parent_timestamp: u64,
     ) -> eyre::Result<()> {
-        // Create channel for transaction batches - fetcher sends CollectionResult
-        let (tx_sender, tx_receiver) = mpsc::channel::<CollectionResult>(self.prefetch_buffer);
+        // Create channel for transaction batches (one batch per payload)
+        let (tx_sender, mut tx_receiver) = mpsc::channel::<Vec<Bytes>>(self.prefetch_buffer);
 
         // Spawn background task to continuously fetch transaction batches
         let rpc_url = self.rpc_url.clone();
         let target_gas = self.target_gas;
+        let count = self.count;
 
         let fetcher_handle = tokio::spawn(async move {
             let tx_source = match RpcTransactionSource::from_url(&rpc_url) {
                 Ok(source) => source,
                 Err(e) => {
                     warn!(error = %e, "Failed to create transaction source");
-                    return None;
+                    return;
                 }
             };
 
             let collector = TransactionCollector::new(tx_source, target_gas);
             let mut current_block = start_block;
 
-            while let Some(batch) = fetch_batch_with_retry(&collector, current_block).await {
-                if batch.transactions.is_empty() {
-                    info!(block = current_block, "Reached chain tip, stopping fetcher");
+            for payload_idx in 0..count {
+                const MAX_RETRIES: u32 = 5;
+                let mut attempts = 0;
+                let result = loop {
+                    attempts += 1;
+                    match collector.collect(current_block).await {
+                        Ok(res) => break Some(res),
+                        Err(e) => {
+                            if attempts >= MAX_RETRIES {
+                                warn!(payload = payload_idx + 1, attempts, error = %e, "Failed to fetch transactions after max retries");
+                                break None;
+                            }
+                            warn!(payload = payload_idx + 1, attempts, error = %e, "Failed to fetch transactions, retrying...");
+                            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                        }
+                    }
+                };
+
+                let Some((transactions, total_gas, next_block)) = result else {
                     break;
-                }
+                };
 
                 info!(
-                    tx_count = batch.transactions.len(),
-                    gas_sent = batch.gas_sent,
-                    blocks = format!("{}..{}", current_block, batch.next_block),
-                    "Fetched transaction batch"
+                    payload = payload_idx + 1,
+                    tx_count = transactions.len(),
+                    total_gas,
+                    blocks = format!("{}..{}", current_block, next_block),
+                    "Fetched transactions"
                 );
-                current_block = batch.next_block;
+                current_block = next_block;
 
-                if tx_sender.send(batch).await.is_err() {
+                if tx_sender.send(transactions).await.is_err() {
                     break;
                 }
             }
-
-            Some(current_block)
         });
 
-        // Transaction buffer: holds transactions from batches + any extras from retries
-        let mut tx_buffer = TxBuffer::new(tx_receiver);
-
         let mut parent_hash = initial_parent_hash;
         let mut parent_timestamp = initial_parent_timestamp;
+        let mut pending_build: Option<tokio::task::JoinHandle<eyre::Result<BuiltPayload>>> = None;
 
         for i in 0..self.count {
-            // Get initial batch of transactions for this payload
-            let mut result = tx_buffer
-                .take_batch()
-                .await
-                .ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
+            let is_last = i == self.count - 1;
 
-            if result.transactions.is_empty() {
-                return Err(eyre::eyre!("No transactions collected for payload {}", i + 1));
-            }
+            // Get current payload (either from pending build or build now)
+            let current_payload = if let Some(handle) = pending_build.take() {
+                handle.await??
+            } else {
+                // First payload - wait for transactions and build synchronously
+                let transactions = tx_receiver
+                    .recv()
+                    .await
+                    .ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
 
-            // Build with retry - may need to request more transactions
-            let built = self
-                .build_with_retry_buffered(
+                if transactions.is_empty() {
+                    return Err(eyre::eyre!("No transactions collected for payload {}", i + 1));
+                }
+
+                info!(
+                    payload = i + 1,
+                    total = self.count,
+                    parent_hash = %parent_hash,
+                    parent_timestamp = parent_timestamp,
+                    tx_count = transactions.len(),
+                    "Building payload via testing_buildBlockV1"
+                );
+                self.build_payload(
                     testing_provider,
-                    &mut tx_buffer,
-                    &mut result,
+                    &transactions,
                     i,
                     parent_hash,
                     parent_timestamp,
                 )
-                .await?;
+                .await?
+            };
 
-            self.save_payload(&built)?;
+            self.save_payload(&current_payload)?;
 
-            let current_block_hash = built.block_hash;
-            let current_timestamp = built.timestamp;
+            let current_block_hash = current_payload.block_hash;
+            let current_timestamp = current_payload.timestamp;
 
-            // Execute payload
-            info!(payload = i + 1, block_hash = %current_block_hash, gas_used = built.gas_used, "Executing payload (newPayload + FCU)");
-            self.execute_payload_v4(auth_provider, built.envelope, parent_hash).await?;
+            // Execute current payload first
+            info!(payload = i + 1, block_hash = %current_block_hash, "Executing payload (newPayload + FCU)");
+            self.execute_payload_v4(auth_provider, current_payload.envelope, parent_hash).await?;
             info!(payload = i + 1, "Payload executed successfully");
 
+            // Start building next payload in background (if not last) - AFTER execution
+            if !is_last {
+                // Get transactions for next payload (should already be fetched or fetching)
+                let next_transactions = tx_receiver
+                    .recv()
+                    .await
+                    .ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
+
+                if next_transactions.is_empty() {
+                    return Err(eyre::eyre!("No transactions collected for payload {}", i + 2));
+                }
+
+                let testing_provider = testing_provider.clone();
+                let next_index = i + 1;
+                let total = self.count;
+
+                pending_build = Some(tokio::spawn(async move {
+                    info!(
+                        payload = next_index + 1,
+                        total = total,
+                        parent_hash = %current_block_hash,
+                        parent_timestamp = current_timestamp,
+                        tx_count = next_transactions.len(),
+                        "Building payload via testing_buildBlockV1"
+                    );
+
+                    Self::build_payload_static(
+                        &testing_provider,
+                        &next_transactions,
+                        next_index,
+                        current_block_hash,
+                        current_timestamp,
+                    )
+                    .await
+                }));
+            }
+
             parent_hash = current_block_hash;
             parent_timestamp = current_timestamp;
         }
 
         // Clean up the fetcher task
-        drop(tx_buffer);
+        drop(tx_receiver);
         let _ = fetcher_handle.await;
 
         Ok(())
     }
 
-    /// Build a payload with retry logic, using the buffered transaction source.
-    async fn build_with_retry_buffered(
+    /// Build a single payload via `testing_buildBlockV1`.
+    async fn build_payload(
         &self,
         testing_provider: &RootProvider<AnyNetwork>,
-        tx_buffer: &mut TxBuffer,
-        result: &mut CollectionResult,
+        transactions: &[Bytes],
         index: u64,
         parent_hash: B256,
         parent_timestamp: u64,
     ) -> eyre::Result<BuiltPayload> {
-        for attempt in 1..=MAX_BUILD_RETRIES {
-            let tx_bytes: Vec<Bytes> = result.transactions.iter().map(|t| t.raw.clone()).collect();
-            let gas_sent = result.gas_sent;
-
-            info!(
-                payload = index + 1,
-                attempt,
-                tx_count = tx_bytes.len(),
-                gas_sent,
-                parent_hash = %parent_hash,
-                "Building payload via testing_buildBlockV1"
-            );
-
-            let built = Self::build_payload_static(
-                testing_provider,
-                &tx_bytes,
-                index,
-                parent_hash,
-                parent_timestamp,
-            )
-            .await?;
-
-            match self.check_retry_outcome(&built, index, attempt, gas_sent) {
-                RetryOutcome::Success | RetryOutcome::MaxRetries => return Ok(built),
-                RetryOutcome::NeedMore(additional_gas) => {
-                    let mut collected_gas = 0u64;
-                    while collected_gas < additional_gas {
-                        if let Some(batch) = tx_buffer.take_batch().await {
-                            collected_gas += batch.gas_sent;
-                            result.transactions.extend(batch.transactions);
-                            result.gas_sent = result.gas_sent.saturating_add(batch.gas_sent);
-                            result.next_block = batch.next_block;
-                        } else {
-                            warn!("Transaction fetcher exhausted, proceeding with available transactions");
-                            break;
-                        }
-                    }
-                }
-            }
-        }
-
-        warn!(payload = index + 1, "Retry loop exited without returning a payload");
-        Err(eyre::eyre!("build_with_retry_buffered exhausted retries without result"))
+        Self::build_payload_static(
+            testing_provider,
+            transactions,
+            index,
+            parent_hash,
+            parent_timestamp,
+        )
+        .await
     }
 
-    /// Determines the outcome of a build attempt.
-    fn check_retry_outcome(
-        &self,
-        built: &BuiltPayload,
-        index: u64,
-        attempt: u32,
-        gas_sent: u64,
-    ) -> RetryOutcome {
-        let gas_used = built.gas_used;
-
-        if gas_used + MIN_TARGET_SLACK >= self.target_gas {
-            info!(
-                payload = index + 1,
-                gas_used,
-                target_gas = self.target_gas,
-                attempts = attempt,
-                "Payload built successfully"
-            );
-            return RetryOutcome::Success;
-        }
-
-        if attempt == MAX_BUILD_RETRIES {
-            warn!(
-                payload = index + 1,
-                gas_used,
-                target_gas = self.target_gas,
-                gas_sent,
-                "Underfilled after max retries, accepting payload"
-            );
-            return RetryOutcome::MaxRetries;
-        }
-
-        if gas_used == 0 {
-            warn!(
-                payload = index + 1,
-                "Zero gas used in payload, requesting fixed chunk of additional transactions"
-            );
-            return RetryOutcome::NeedMore(self.target_gas);
-        }
-
-        let gas_sent_needed_total =
-            (self.target_gas as u128 * gas_sent as u128).div_ceil(gas_used as u128) as u64;
-        let additional = gas_sent_needed_total.saturating_sub(gas_sent);
-        let additional = additional.min(self.target_gas * MAX_ADDITIONAL_GAS_MULTIPLIER);
-
-        if additional == 0 {
-            info!(
-                payload = index + 1,
-                gas_used,
-                target_gas = self.target_gas,
-                "No additional transactions needed based on ratio"
-            );
-            return RetryOutcome::Success;
-        }
-
-        let ratio = gas_used as f64 / gas_sent as f64;
-        info!(
-            payload = index + 1,
-            gas_used,
-            gas_sent,
-            ratio = format!("{:.4}", ratio),
-            additional_gas = additional,
-            "Underfilled, collecting more transactions for retry"
-        );
-        RetryOutcome::NeedMore(additional)
-    }
-
-    /// Build a single payload via `testing_buildBlockV1`.
+    /// Static version for use in spawned tasks.
     async fn build_payload_static(
         testing_provider: &RootProvider<AnyNetwork>,
         transactions: &[Bytes],
@@ -774,9 +591,8 @@ impl Command {
         let block_hash = inner.block_hash;
         let block_number = inner.block_number;
         let timestamp = inner.timestamp;
-        let gas_used = inner.gas_used;
 
-        Ok(BuiltPayload { block_number, envelope: v4_envelope, block_hash, timestamp, gas_used })
+        Ok(BuiltPayload { block_number, envelope: v4_envelope, block_hash, timestamp })
     }
 
     /// Save a payload to disk.

bin/reth-bench/src/bench/helpers.rs

@@ -1,33 +1,6 @@
 //! Common helpers for reth-bench commands.
 
 use crate::valid_payload::call_forkchoice_updated;
-use eyre::Result;
-use std::io::{BufReader, Read};
-
-/// Read input from either a file path or stdin.
-pub(crate) fn read_input(path: Option<&str>) -> Result<String> {
-    Ok(match path {
-        Some(path) => reth_fs_util::read_to_string(path)?,
-        None => String::from_utf8(
-            BufReader::new(std::io::stdin()).bytes().collect::<Result<Vec<_>, _>>()?,
-        )?,
-    })
-}
-
-/// Load JWT secret from either a file or use the provided string directly.
-pub(crate) fn load_jwt_secret(jwt_secret: Option<&str>) -> Result<Option<String>> {
-    match jwt_secret {
-        Some(secret) => {
-            // Try to read as file first
-            match std::fs::read_to_string(secret) {
-                Ok(contents) => Ok(Some(contents.trim().to_string())),
-                // If file read fails, use the string directly
-                Err(_) => Ok(Some(secret.to_string())),
-            }
-        }
-        None => Ok(None),
-    }
-}
 
 /// Parses a gas limit value with optional suffix: K for thousand, M for million, G for billion.
 ///

bin/reth-bench/src/bench/new_payload_fcu.rs

@@ -16,7 +16,7 @@ use crate::{
             write_benchmark_results, CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow,
         },
         persistence_waiter::{
-            derive_ws_rpc_url, setup_persistence_subscription, PersistenceWaiter,
+            engine_url_to_ws_url, setup_persistence_subscription, PersistenceWaiter,
             PERSISTENCE_CHECKPOINT_TIMEOUT,
         },
     },
@@ -32,6 +32,7 @@ use reth_engine_primitives::config::DEFAULT_PERSISTENCE_THRESHOLD;
 use reth_node_core::args::BenchmarkArgs;
 use std::time::{Duration, Instant};
 use tracing::{debug, info};
+use url::Url;
 
 /// `reth benchmark new-payload-fcu` command
 #[derive(Debug, Parser)]
@@ -100,10 +101,7 @@ impl Command {
         let mut waiter = match (self.wait_time, self.wait_for_persistence) {
             (Some(duration), _) => Some(PersistenceWaiter::with_duration(duration)),
             (None, true) => {
-                let ws_url = derive_ws_rpc_url(
-                    self.benchmark.ws_rpc_url.as_deref(),
-                    &self.benchmark.engine_rpc_url,
-                )?;
+                let ws_url = self.derive_ws_rpc_url()?;
                 let sub = setup_persistence_subscription(ws_url).await?;
                 Some(PersistenceWaiter::with_subscription(
                     sub,
@@ -262,4 +260,33 @@ impl Command {
 
         Ok(())
     }
+
+    /// Returns the websocket RPC URL used for the persistence subscription.
+    ///
+    /// Preference:
+    /// - If `--ws-rpc-url` is provided, use it directly.
+    /// - Otherwise, derive a WS RPC URL from `--engine-rpc-url`.
+    ///
+    /// The persistence subscription endpoint (`reth_subscribePersistedBlock`) is exposed on
+    /// the regular RPC server (WS port, usually 8546), not on the engine API port (usually 8551).
+    /// Since `BenchmarkArgs` only has the engine URL by default, we convert the scheme
+    /// (http→ws, https→wss) and force the port to 8546.
+    fn derive_ws_rpc_url(&self) -> eyre::Result<Url> {
+        if let Some(ref ws_url) = self.benchmark.ws_rpc_url {
+            let parsed: Url = ws_url
+                .parse()
+                .wrap_err_with(|| format!("Failed to parse WebSocket RPC URL: {ws_url}"))?;
+            info!(target: "reth-bench", ws_url = %parsed, "Using provided WebSocket RPC URL");
+            Ok(parsed)
+        } else {
+            let derived = engine_url_to_ws_url(&self.benchmark.engine_rpc_url)?;
+            debug!(
+                target: "reth-bench",
+                engine_url = %self.benchmark.engine_rpc_url,
+                %derived,
+                "Derived WebSocket RPC URL from engine RPC URL"
+            );
+            Ok(derived)
+        }
+    }
 }

bin/reth-bench/src/bench/persistence_waiter.rs

@@ -15,46 +15,11 @@ use eyre::Context;
 use futures::StreamExt;
 use std::time::Duration;
 use tracing::{debug, info};
-
-/// Default `WebSocket` RPC port for reth.
-const DEFAULT_WS_RPC_PORT: u16 = 8546;
 use url::Url;
 
 /// Default timeout for waiting on persistence.
 pub(crate) const PERSISTENCE_CHECKPOINT_TIMEOUT: Duration = Duration::from_secs(60);
 
-/// Returns the websocket RPC URL used for the persistence subscription.
-///
-/// Preference:
-/// - If `ws_rpc_url` is provided, use it directly.
-/// - Otherwise, derive a WS RPC URL from `engine_rpc_url`.
-///
-/// The persistence subscription endpoint (`reth_subscribePersistedBlock`) is exposed on
-/// the regular RPC server (WS port, usually 8546), not on the engine API port (usually 8551).
-/// Since we may only have the engine URL by default, we convert the scheme
-/// (http→ws, https→wss) and force the port to 8546.
-pub(crate) fn derive_ws_rpc_url(
-    ws_rpc_url: Option<&str>,
-    engine_rpc_url: &str,
-) -> eyre::Result<Url> {
-    if let Some(ws_url) = ws_rpc_url {
-        let parsed: Url = ws_url
-            .parse()
-            .wrap_err_with(|| format!("Failed to parse WebSocket RPC URL: {ws_url}"))?;
-        info!(target: "reth-bench", ws_url = %parsed, "Using provided WebSocket RPC URL");
-        Ok(parsed)
-    } else {
-        let derived = engine_url_to_ws_url(engine_rpc_url)?;
-        debug!(
-            target: "reth-bench",
-            engine_url = %engine_rpc_url,
-            %derived,
-            "Derived WebSocket RPC URL from engine RPC URL"
-        );
-        Ok(derived)
-    }
-}
-
 /// Converts an engine API URL to the default RPC websocket URL.
 ///
 /// Transformations:
@@ -65,7 +30,7 @@ pub(crate) fn derive_ws_rpc_url(
 ///
 /// This is used when we only know the engine API URL (typically `:8551`) but
 /// need to connect to the node's WS RPC endpoint for persistence events.
-fn engine_url_to_ws_url(engine_url: &str) -> eyre::Result<Url> {
+pub(crate) fn engine_url_to_ws_url(engine_url: &str) -> eyre::Result<Url> {
     let url: Url = engine_url
         .parse()
         .wrap_err_with(|| format!("Failed to parse engine RPC URL: {engine_url}"))?;
@@ -87,9 +52,7 @@ fn engine_url_to_ws_url(engine_url: &str) -> eyre::Result<Url> {
         }
     }
 
-    ws_url
-        .set_port(Some(DEFAULT_WS_RPC_PORT))
-        .map_err(|_| eyre::eyre!("Failed to set port for URL: {url}"))?;
+    ws_url.set_port(Some(8546)).map_err(|_| eyre::eyre!("Failed to set port for URL: {url}"))?;
 
     Ok(ws_url)
 }

bin/reth-bench/src/bench/replay_payloads.rs

@@ -19,7 +19,7 @@ use crate::{
             TotalGasOutput, TotalGasRow,
         },
         persistence_waiter::{
-            derive_ws_rpc_url, setup_persistence_subscription, PersistenceWaiter,
+            engine_url_to_ws_url, setup_persistence_subscription, PersistenceWaiter,
             PERSISTENCE_CHECKPOINT_TIMEOUT,
         },
     },
@@ -152,7 +152,7 @@ impl Command {
         let mut waiter = match (self.wait_time, self.wait_for_persistence) {
             (Some(duration), _) => Some(PersistenceWaiter::with_duration(duration)),
             (None, true) => {
-                let ws_url = derive_ws_rpc_url(self.ws_rpc_url.as_deref(), &self.engine_rpc_url)?;
+                let ws_url = self.derive_ws_rpc_url()?;
                 let sub = setup_persistence_subscription(ws_url).await?;
                 Some(PersistenceWaiter::with_subscription(
                     sub,
@@ -464,4 +464,33 @@ impl Command {
 
         Ok(payloads)
     }
+
+    /// Returns the websocket RPC URL used for the persistence subscription.
+    ///
+    /// Preference:
+    /// - If `--ws-rpc-url` is provided, use it directly.
+    /// - Otherwise, derive a WS RPC URL from `--engine-rpc-url`.
+    ///
+    /// The persistence subscription endpoint (`reth_subscribePersistedBlock`) is exposed on
+    /// the regular RPC server (WS port, usually 8546), not on the engine API port (usually 8551).
+    /// Since we only have the engine URL by default, we convert the scheme
+    /// (http→ws, https→wss) and force the port to 8546.
+    fn derive_ws_rpc_url(&self) -> eyre::Result<Url> {
+        if let Some(ref ws_url) = self.ws_rpc_url {
+            let parsed: Url = ws_url
+                .parse()
+                .wrap_err_with(|| format!("Failed to parse WebSocket RPC URL: {ws_url}"))?;
+            info!(target: "reth-bench", ws_url = %parsed, "Using provided WebSocket RPC URL");
+            Ok(parsed)
+        } else {
+            let derived = engine_url_to_ws_url(&self.engine_rpc_url)?;
+            debug!(
+                target: "reth-bench",
+                engine_url = %self.engine_rpc_url,
+                %derived,
+                "Derived WebSocket RPC URL from engine RPC URL"
+            );
+            Ok(derived)
+        }
+    }
 }

bin/reth-bench/src/bench/send_invalid_payload/mod.rs

@@ -3,7 +3,6 @@
 mod invalidation;
 use invalidation::InvalidationConfig;
 
-use super::helpers::{load_jwt_secret, read_input};
 use alloy_primitives::{Address, B256};
 use alloy_provider::network::AnyRpcBlock;
 use alloy_rpc_types_engine::ExecutionPayload;
@@ -11,7 +10,7 @@ use clap::Parser;
 use eyre::{OptionExt, Result};
 use op_alloy_consensus::OpTxEnvelope;
 use reth_cli_runner::CliContext;
-use std::io::Write;
+use std::io::{BufReader, Read, Write};
 
 /// Command for generating and sending an invalid `engine_newPayload` request.
 ///
@@ -181,6 +180,27 @@ enum Mode {
 }
 
 impl Command {
+    /// Read input from either a file or stdin
+    fn read_input(&self) -> Result<String> {
+        Ok(match &self.path {
+            Some(path) => reth_fs_util::read_to_string(path)?,
+            None => String::from_utf8(
+                BufReader::new(std::io::stdin()).bytes().collect::<Result<Vec<_>, _>>()?,
+            )?,
+        })
+    }
+
+    /// Load JWT secret from either a file or use the provided string directly
+    fn load_jwt_secret(&self) -> Result<Option<String>> {
+        match &self.jwt_secret {
+            Some(secret) => match std::fs::read_to_string(secret) {
+                Ok(contents) => Ok(Some(contents.trim().to_string())),
+                Err(_) => Ok(Some(secret.clone())),
+            },
+            None => Ok(None),
+        }
+    }
+
     /// Build `InvalidationConfig` from command flags
     const fn build_invalidation_config(&self) -> InvalidationConfig {
         InvalidationConfig {
@@ -216,8 +236,8 @@ impl Command {
 
     /// Execute the command
     pub async fn execute(self, _ctx: CliContext) -> Result<()> {
-        let block_json = read_input(self.path.as_deref())?;
-        let jwt_secret = load_jwt_secret(self.jwt_secret.as_deref())?;
+        let block_json = self.read_input()?;
+        let jwt_secret = self.load_jwt_secret()?;
 
         let block = serde_json::from_str::<AnyRpcBlock>(&block_json)?
             .into_inner()

bin/reth-bench/src/bench/send_payload.rs

@@ -1,11 +1,10 @@
-use super::helpers::{load_jwt_secret, read_input};
 use alloy_provider::network::AnyRpcBlock;
 use alloy_rpc_types_engine::ExecutionPayload;
 use clap::Parser;
 use eyre::{OptionExt, Result};
 use op_alloy_consensus::OpTxEnvelope;
 use reth_cli_runner::CliContext;
-use std::io::Write;
+use std::io::{BufReader, Read, Write};
 
 /// Command for generating and sending an `engine_newPayload` request constructed from an RPC
 /// block.
@@ -52,13 +51,38 @@ enum Mode {
 }
 
 impl Command {
+    /// Read input from either a file or stdin
+    fn read_input(&self) -> Result<String> {
+        Ok(match &self.path {
+            Some(path) => reth_fs_util::read_to_string(path)?,
+            None => String::from_utf8(
+                BufReader::new(std::io::stdin()).bytes().collect::<Result<Vec<_>, _>>()?,
+            )?,
+        })
+    }
+
+    /// Load JWT secret from either a file or use the provided string directly
+    fn load_jwt_secret(&self) -> Result<Option<String>> {
+        match &self.jwt_secret {
+            Some(secret) => {
+                // Try to read as file first
+                match std::fs::read_to_string(secret) {
+                    Ok(contents) => Ok(Some(contents.trim().to_string())),
+                    // If file read fails, use the string directly
+                    Err(_) => Ok(Some(secret.clone())),
+                }
+            }
+            None => Ok(None),
+        }
+    }
+
     /// Execute the generate payload command
     pub async fn execute(self, _ctx: CliContext) -> Result<()> {
         // Load block
-        let block_json = read_input(self.path.as_deref())?;
+        let block_json = self.read_input()?;
 
         // Load JWT secret
-        let jwt_secret = load_jwt_secret(self.jwt_secret.as_deref())?;
+        let jwt_secret = self.load_jwt_secret()?;
 
         // Parse the block
         let block = serde_json::from_str::<AnyRpcBlock>(&block_json)?

crates/chainspec/src/spec.rs

@@ -541,7 +541,7 @@ impl<H: BlockHeader> ChainSpec<H> {
                     }
                 }
 
-                bf_params.first().map(|(_, params)| *params).unwrap_or_else(BaseFeeParams::ethereum)
+                bf_params.first().map(|(_, params)| *params).unwrap_or(BaseFeeParams::ethereum())
             }
         }
     }

crates/cli/commands/Cargo.toml

@@ -78,7 +78,6 @@ lz4.workspace = true
 zstd.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-parking_lot.workspace = true
 tar.workspace = true
 tracing.workspace = true
 backon.workspace = true

crates/cli/commands/src/common.rs

@@ -121,13 +121,13 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         let genesis_block_number = self.chain.genesis().number.unwrap_or_default();
         let (db, sfp) = match access {
             AccessRights::RW => (
-                init_db(db_path, self.db.database_args())?,
+                Arc::new(init_db(db_path, self.db.database_args())?),
                 StaticFileProviderBuilder::read_write(sf_path)
                     .with_genesis_block_number(genesis_block_number)
                     .build()?,
             ),
             AccessRights::RO | AccessRights::RoInconsistent => {
-                (open_db_read_only(&db_path, self.db.database_args())?, {
+                (Arc::new(open_db_read_only(&db_path, self.db.database_args())?), {
                     let provider = StaticFileProviderBuilder::read_only(sf_path)
                         .with_genesis_block_number(genesis_block_number)
                         .build()?;
@@ -160,16 +160,16 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
     fn create_provider_factory<N: CliNodeTypes>(
         &self,
         config: &Config,
-        db: DatabaseEnv,
+        db: Arc<DatabaseEnv>,
         static_file_provider: StaticFileProvider<N::Primitives>,
         rocksdb_provider: RocksDBProvider,
         access: AccessRights,
-    ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, DatabaseEnv>>>
+    ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>>
     where
         C: ChainSpecParser<ChainSpec = N::ChainSpec>,
     {
         let prune_modes = config.prune.segments.clone();
-        let factory = ProviderFactory::<NodeTypesWithDBAdapter<N, DatabaseEnv>>::new(
+        let factory = ProviderFactory::<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>::new(
             db,
             self.chain.clone(),
             static_file_provider,
@@ -200,7 +200,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
             let (_tip_tx, tip_rx) = watch::channel(B256::ZERO);
 
             // Builds and executes an unwind-only pipeline
-            let mut pipeline = Pipeline::<NodeTypesWithDBAdapter<N, DatabaseEnv>>::builder()
+            let mut pipeline = Pipeline::<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>::builder()
                 .add_stages(DefaultStages::new(
                     factory.clone(),
                     tip_rx,
@@ -229,7 +229,7 @@ pub struct Environment<N: NodeTypes> {
     /// Configuration for reth node
     pub config: Config,
     /// Provider factory.
-    pub provider_factory: ProviderFactory<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+    pub provider_factory: ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     /// Datadir path.
     pub data_dir: ChainPath<DataDirPath>,
 }
@@ -261,8 +261,8 @@ impl AccessRights {
 /// Helper alias to satisfy `FullNodeTypes` bound on [`Node`] trait generic.
 type FullTypesAdapter<T> = FullNodeTypesAdapter<
     T,
-    DatabaseEnv,
-    BlockchainProvider<NodeTypesWithDBAdapter<T, DatabaseEnv>>,
+    Arc<DatabaseEnv>,
+    BlockchainProvider<NodeTypesWithDBAdapter<T, Arc<DatabaseEnv>>>,
 >;
 
 /// Helper trait with a common set of requirements for the

crates/cli/commands/src/db/checksum/mod.rs

@@ -17,6 +17,7 @@ use reth_provider::{providers::ProviderNodeTypes, DBProvider, StaticFileProvider
 use reth_static_file_types::StaticFileSegment;
 use std::{
     hash::{BuildHasher, Hasher},
+    sync::Arc,
     time::{Duration, Instant},
 };
 use tracing::{info, warn};
@@ -89,7 +90,7 @@ impl Command {
     /// Execute `db checksum` command
     pub fn execute<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
         self,
-        tool: &DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+        tool: &DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     ) -> eyre::Result<()> {
         warn!("This command should be run without the node running!");
 
@@ -116,7 +117,7 @@ fn checksum_hasher() -> impl Hasher {
 }
 
 fn checksum_static_file<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
-    tool: &DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+    tool: &DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     segment: StaticFileSegment,
     start_block: Option<u64>,
     end_block: Option<u64>,

crates/cli/commands/src/db/checksum/rocksdb.rs

@@ -9,7 +9,7 @@ use reth_db_api::table::Table;
 use reth_db_common::DbTool;
 use reth_node_builder::NodeTypesWithDBAdapter;
 use reth_provider::RocksDBProviderFactory;
-use std::{hash::Hasher, time::Instant};
+use std::{hash::Hasher, sync::Arc, time::Instant};
 use tracing::info;
 
 /// RocksDB tables that can be checksummed.
@@ -36,7 +36,7 @@ impl RocksDbTable {
 
 /// Computes a checksum for a RocksDB table.
 pub fn checksum_rocksdb<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
-    tool: &DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+    tool: &DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     table: RocksDbTable,
     limit: Option<usize>,
 ) -> eyre::Result<()> {

crates/cli/commands/src/db/diff.rs

@@ -16,6 +16,7 @@ use std::{
     hash::Hash,
     io::Write,
     path::{Path, PathBuf},
+    sync::Arc,
 };
 use tracing::{info, warn};
 
@@ -55,7 +56,7 @@ impl Command {
     /// then written to a file in the output directory.
     pub fn execute<T: NodeTypes>(
         self,
-        tool: &DbTool<NodeTypesWithDBAdapter<T, DatabaseEnv>>,
+        tool: &DbTool<NodeTypesWithDBAdapter<T, Arc<DatabaseEnv>>>,
     ) -> eyre::Result<()> {
         warn!("Make sure the node is not running when running `reth db diff`!");
         // open second db

crates/cli/commands/src/db/list.rs

@@ -7,7 +7,7 @@ use reth_db::{transaction::DbTx, DatabaseEnv};
 use reth_db_api::{database::Database, table::Table, RawValue, TableViewer, Tables};
 use reth_db_common::{DbTool, ListFilter};
 use reth_node_builder::{NodeTypes, NodeTypesWithDBAdapter};
-use std::cell::RefCell;
+use std::{cell::RefCell, sync::Arc};
 use tracing::error;
 
 #[derive(Parser, Debug)]
@@ -55,7 +55,7 @@ impl Command {
     /// Execute `db list` command
     pub fn execute<N: NodeTypes<ChainSpec: EthereumHardforks>>(
         self,
-        tool: &DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+        tool: &DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     ) -> eyre::Result<()> {
         self.table.view(&ListTableViewer { tool, args: &self })
     }
@@ -89,7 +89,7 @@ impl Command {
 }
 
 struct ListTableViewer<'a, N: NodeTypes> {
-    tool: &'a DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+    tool: &'a DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     args: &'a Command,
 }
 

crates/cli/commands/src/db/mod.rs

@@ -17,7 +17,6 @@ mod get;
 mod list;
 mod repair_trie;
 mod settings;
-mod state;
 mod static_file_header;
 mod stats;
 /// DB List TUI
@@ -66,8 +65,6 @@ pub enum Subcommands {
     Settings(settings::Command),
     /// Gets storage size information for an account
     AccountStorage(account_storage::Command),
-    /// Gets account state and storage at a specific block
-    State(state::Command),
 }
 
 /// Initializes a provider factory with specified access rights, and then execute with the provided
@@ -201,11 +198,6 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
                     command.execute(&tool)?;
                 });
             }
-            Subcommands::State(command) => {
-                db_exec!(self.env, tool, N, AccessRights::RO, {
-                    command.execute(&tool)?;
-                });
-            }
         }
 
         Ok(())

crates/cli/commands/src/db/state.rs

@@ -1,412 +0,0 @@
-use alloy_primitives::{Address, BlockNumber, B256, U256};
-use clap::Parser;
-use parking_lot::Mutex;
-use reth_db_api::{
-    cursor::{DbCursorRO, DbDupCursorRO},
-    database::Database,
-    tables,
-    transaction::DbTx,
-};
-use reth_db_common::DbTool;
-use reth_node_builder::NodeTypesWithDB;
-use reth_provider::providers::ProviderNodeTypes;
-use reth_storage_api::{BlockNumReader, StateProvider, StorageSettingsCache};
-use std::{
-    collections::BTreeSet,
-    thread,
-    time::{Duration, Instant},
-};
-use tracing::{error, info};
-
-/// Log progress every 5 seconds
-const LOG_INTERVAL: Duration = Duration::from_secs(30);
-
-/// The arguments for the `reth db state` command
-#[derive(Parser, Debug)]
-pub struct Command {
-    /// The account address to get state for
-    address: Address,
-
-    /// Block number to query state at (uses current state if not provided)
-    #[arg(long, short)]
-    block: Option<BlockNumber>,
-
-    /// Maximum number of storage slots to display
-    #[arg(long, short, default_value = "100")]
-    limit: usize,
-
-    /// Output format (table, json, csv)
-    #[arg(long, short, default_value = "table")]
-    format: OutputFormat,
-}
-
-impl Command {
-    /// Execute `db state` command
-    pub fn execute<N: NodeTypesWithDB + ProviderNodeTypes>(
-        self,
-        tool: &DbTool<N>,
-    ) -> eyre::Result<()> {
-        let address = self.address;
-        let limit = self.limit;
-
-        if let Some(block) = self.block {
-            self.execute_historical(tool, address, block, limit)
-        } else {
-            self.execute_current(tool, address, limit)
-        }
-    }
-
-    fn execute_current<N: NodeTypesWithDB + ProviderNodeTypes>(
-        &self,
-        tool: &DbTool<N>,
-        address: Address,
-        limit: usize,
-    ) -> eyre::Result<()> {
-        let entries = tool.provider_factory.db_ref().view(|tx| {
-            // Get account info
-            let account = tx.get::<tables::PlainAccountState>(address)?;
-
-            // Get storage entries
-            let mut cursor = tx.cursor_dup_read::<tables::PlainStorageState>()?;
-            let mut entries = Vec::new();
-            let mut last_log = Instant::now();
-
-            let walker = cursor.walk_dup(Some(address), None)?;
-            for (idx, entry) in walker.enumerate() {
-                let (_, storage_entry) = entry?;
-
-                if storage_entry.value != U256::ZERO {
-                    entries.push((storage_entry.key, storage_entry.value));
-                }
-
-                if entries.len() >= limit {
-                    break;
-                }
-
-                if last_log.elapsed() >= LOG_INTERVAL {
-                    info!(
-                        target: "reth::cli",
-                        address = %address,
-                        slots_scanned = idx,
-                        "Scanning storage slots"
-                    );
-                    last_log = Instant::now();
-                }
-            }
-
-            Ok::<_, eyre::Report>((account, entries))
-        })??;
-
-        let (account, storage_entries) = entries;
-
-        self.print_results(address, None, account, &storage_entries);
-
-        Ok(())
-    }
-
-    fn execute_historical<N: NodeTypesWithDB + ProviderNodeTypes>(
-        &self,
-        tool: &DbTool<N>,
-        address: Address,
-        block: BlockNumber,
-        limit: usize,
-    ) -> eyre::Result<()> {
-        let provider = tool.provider_factory.history_by_block_number(block)?;
-
-        // Get account info at that block
-        let account = provider.basic_account(&address)?;
-
-        // Check storage settings to determine where history is stored
-        let storage_settings = tool.provider_factory.cached_storage_settings();
-        let history_in_rocksdb = storage_settings.storages_history_in_rocksdb;
-
-        // For historical queries, enumerate keys from history indices only
-        // (not PlainStorageState, which reflects current state)
-        let mut storage_keys = BTreeSet::new();
-
-        if history_in_rocksdb {
-            error!(
-                target: "reth::cli",
-                "Historical storage queries with RocksDB backend are not yet supported. \
-                 Use MDBX for storage history or query current state without --block."
-            );
-            return Ok(());
-        }
-
-        // Collect keys from MDBX StorageChangeSets using parallel scanning
-        self.collect_mdbx_storage_keys_parallel(tool, address, &mut storage_keys)?;
-
-        info!(
-            target: "reth::cli",
-            address = %address,
-            block = block,
-            total_keys = storage_keys.len(),
-            "Found storage keys to query"
-        );
-
-        // Now query each key at the historical block using the StateProvider
-        // This handles both MDBX and RocksDB backends transparently
-        let mut entries = Vec::new();
-        let mut last_log = Instant::now();
-
-        for (idx, key) in storage_keys.iter().enumerate() {
-            match provider.storage(address, *key) {
-                Ok(Some(value)) if value != U256::ZERO => {
-                    entries.push((*key, value));
-                }
-                _ => {}
-            }
-
-            if entries.len() >= limit {
-                break;
-            }
-
-            if last_log.elapsed() >= LOG_INTERVAL {
-                info!(
-                    target: "reth::cli",
-                    address = %address,
-                    block = block,
-                    keys_total = storage_keys.len(),
-                    slots_scanned = idx,
-                    slots_found = entries.len(),
-                    "Scanning historical storage slots"
-                );
-                last_log = Instant::now();
-            }
-        }
-
-        self.print_results(address, Some(block), account, &entries);
-
-        Ok(())
-    }
-
-    /// Collects storage keys from MDBX StorageChangeSets using parallel block range scanning.
-    fn collect_mdbx_storage_keys_parallel<N: NodeTypesWithDB + ProviderNodeTypes>(
-        &self,
-        tool: &DbTool<N>,
-        address: Address,
-        keys: &mut BTreeSet<B256>,
-    ) -> eyre::Result<()> {
-        const CHUNK_SIZE: u64 = 500_000; // 500k blocks per thread
-        let num_threads = std::thread::available_parallelism()
-            .map(|p| p.get().saturating_sub(1).max(1))
-            .unwrap_or(4);
-
-        // Get the current tip block
-        let tip = tool.provider_factory.provider()?.best_block_number()?;
-
-        if tip == 0 {
-            return Ok(());
-        }
-
-        info!(
-            target: "reth::cli",
-            address = %address,
-            tip,
-            chunk_size = CHUNK_SIZE,
-            num_threads,
-            "Starting parallel MDBX changeset scan"
-        );
-
-        // Shared state for collecting keys
-        let collected_keys: Mutex<BTreeSet<B256>> = Mutex::new(BTreeSet::new());
-        let total_entries_scanned = Mutex::new(0usize);
-
-        // Create chunk ranges
-        let mut chunks: Vec<(u64, u64)> = Vec::new();
-        let mut start = 0u64;
-        while start <= tip {
-            let end = (start + CHUNK_SIZE - 1).min(tip);
-            chunks.push((start, end));
-            start = end + 1;
-        }
-
-        let chunks_ref = &chunks;
-        let next_chunk = Mutex::new(0usize);
-        let next_chunk_ref = &next_chunk;
-        let collected_keys_ref = &collected_keys;
-        let total_entries_ref = &total_entries_scanned;
-
-        thread::scope(|s| {
-            let handles: Vec<_> = (0..num_threads)
-                .map(|thread_id| {
-                    s.spawn(move || {
-                        loop {
-                            // Get next chunk to process
-                            let chunk_idx = {
-                                let mut idx = next_chunk_ref.lock();
-                                if *idx >= chunks_ref.len() {
-                                    return Ok::<_, eyre::Report>(());
-                                }
-                                let current = *idx;
-                                *idx += 1;
-                                current
-                            };
-
-                            let (chunk_start, chunk_end) = chunks_ref[chunk_idx];
-
-                            // Open a new read transaction for this chunk
-                            tool.provider_factory.db_ref().view(|tx| {
-                                tx.disable_long_read_transaction_safety();
-
-                                let mut changeset_cursor =
-                                    tx.cursor_read::<tables::StorageChangeSets>()?;
-                                let start_key =
-                                    reth_db_api::models::BlockNumberAddress((chunk_start, address));
-                                let end_key =
-                                    reth_db_api::models::BlockNumberAddress((chunk_end, address));
-
-                                let mut local_keys = BTreeSet::new();
-                                let mut entries_in_chunk = 0usize;
-
-                                if let Ok(walker) = changeset_cursor.walk_range(start_key..=end_key)
-                                {
-                                    for (block_addr, storage_entry) in walker.flatten() {
-                                        if block_addr.address() == address {
-                                            local_keys.insert(storage_entry.key);
-                                        }
-                                        entries_in_chunk += 1;
-                                    }
-                                }
-
-                                // Merge into global state
-                                collected_keys_ref.lock().extend(local_keys);
-                                *total_entries_ref.lock() += entries_in_chunk;
-
-                                info!(
-                                    target: "reth::cli",
-                                    thread_id,
-                                    chunk_start,
-                                    chunk_end,
-                                    entries_in_chunk,
-                                    "Thread completed chunk"
-                                );
-
-                                Ok::<_, eyre::Report>(())
-                            })??;
-                        }
-                    })
-                })
-                .collect();
-
-            for handle in handles {
-                handle.join().map_err(|_| eyre::eyre!("Thread panicked"))??;
-            }
-
-            Ok::<_, eyre::Report>(())
-        })?;
-
-        let final_keys = collected_keys.into_inner();
-        let total = *total_entries_scanned.lock();
-
-        info!(
-            target: "reth::cli",
-            address = %address,
-            total_entries = total,
-            unique_keys = final_keys.len(),
-            "Finished parallel MDBX changeset scan"
-        );
-
-        keys.extend(final_keys);
-        Ok(())
-    }
-
-    fn print_results(
-        &self,
-        address: Address,
-        block: Option<BlockNumber>,
-        account: Option<reth_primitives_traits::Account>,
-        storage: &[(alloy_primitives::B256, U256)],
-    ) {
-        match self.format {
-            OutputFormat::Table => {
-                println!("Account: {address}");
-                if let Some(b) = block {
-                    println!("Block: {b}");
-                } else {
-                    println!("Block: latest");
-                }
-                println!();
-
-                if let Some(acc) = account {
-                    println!("Nonce: {}", acc.nonce);
-                    println!("Balance: {} wei", acc.balance);
-                    if let Some(code_hash) = acc.bytecode_hash {
-                        println!("Code hash: {code_hash}");
-                    }
-                } else {
-                    println!("Account not found");
-                }
-
-                println!();
-                println!("Storage ({} slots):", storage.len());
-                println!("{:-<130}", "");
-                println!("{:<66} | {:<64}", "Slot", "Value");
-                println!("{:-<130}", "");
-                for (key, value) in storage {
-                    println!("{key} | {value:#066x}");
-                }
-            }
-            OutputFormat::Json => {
-                let output = serde_json::json!({
-                    "address": address.to_string(),
-                    "block": block,
-                    "account": account.map(|a| serde_json::json!({
-                        "nonce": a.nonce,
-                        "balance": a.balance.to_string(),
-                        "code_hash": a.bytecode_hash.map(|h| h.to_string()),
-                    })),
-                    "storage": storage.iter().map(|(k, v)| {
-                        serde_json::json!({
-                            "key": k.to_string(),
-                            "value": format!("{v:#066x}"),
-                        })
-                    }).collect::<Vec<_>>(),
-                });
-                println!("{}", serde_json::to_string_pretty(&output).unwrap());
-            }
-            OutputFormat::Csv => {
-                println!("slot,value");
-                for (key, value) in storage {
-                    println!("{key},{value:#066x}");
-                }
-            }
-        }
-    }
-}
-
-#[derive(Debug, Clone, Default, clap::ValueEnum)]
-pub enum OutputFormat {
-    #[default]
-    Table,
-    Json,
-    Csv,
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn parse_state_args() {
-        let cmd = Command::try_parse_from([
-            "state",
-            "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
-            "--block",
-            "1000000",
-        ])
-        .unwrap();
-        assert_eq!(
-            cmd.address,
-            "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045".parse::<Address>().unwrap()
-        );
-        assert_eq!(cmd.block, Some(1000000));
-    }
-
-    #[test]
-    fn parse_state_args_no_block() {
-        let cmd = Command::try_parse_from(["state", "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"])
-            .unwrap();
-        assert_eq!(cmd.block, None);
-    }
-}

crates/cli/commands/src/db/stats.rs

@@ -16,7 +16,7 @@ use reth_provider::{
     RocksDBProviderFactory,
 };
 use reth_static_file_types::SegmentRangeInclusive;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 #[derive(Parser, Debug)]
 /// The arguments for the `reth db stats` command
@@ -48,7 +48,7 @@ impl Command {
     pub fn execute<N: CliNodeTypes<ChainSpec: EthereumHardforks>>(
         self,
         data_dir: ChainPath<DataDirPath>,
-        tool: &DbTool<NodeTypesWithDBAdapter<N, DatabaseEnv>>,
+        tool: &DbTool<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
     ) -> eyre::Result<()> {
         if self.checksum {
             let checksum_report = self.checksum_report(tool)?;
@@ -72,7 +72,7 @@ impl Command {
         Ok(())
     }
 
-    fn db_stats_table<N: NodeTypesWithDB<DB = DatabaseEnv>>(
+    fn db_stats_table<N: NodeTypesWithDB<DB = Arc<DatabaseEnv>>>(
         &self,
         tool: &DbTool<N>,
     ) -> eyre::Result<ComfyTable> {

crates/cli/commands/src/db/tui.rs

@@ -227,9 +227,8 @@ where
 
         // Handle errors
         if let Err(err) = res {
-            error!("{err}");
+            error!("{:?}", err)
         }
-
         Ok(())
     }
 }
@@ -242,7 +241,6 @@ fn event_loop<B: Backend, F, T: Table>(
 ) -> io::Result<()>
 where
     F: FnMut(usize, usize) -> Vec<TableRow<T>>,
-    io::Error: From<B::Error>,
 {
     let mut last_tick = Instant::now();
     let mut running = true;

crates/cli/commands/src/launcher.rs

@@ -2,7 +2,7 @@ use futures::Future;
 use reth_cli::chainspec::ChainSpecParser;
 use reth_db::DatabaseEnv;
 use reth_node_builder::{NodeBuilder, WithLaunchContext};
-use std::fmt;
+use std::{fmt, sync::Arc};
 
 /// A trait for launching a reth node with custom configuration strategies.
 ///
@@ -30,7 +30,7 @@ where
     /// * `builder_args` - Extension arguments for configuration
     fn entrypoint(
         self,
-        builder: WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+        builder: WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
         builder_args: Ext,
     ) -> impl Future<Output = eyre::Result<()>>;
 }
@@ -58,7 +58,7 @@ impl<F> FnLauncher<F> {
     where
         C: ChainSpecParser,
         F: AsyncFnOnce(
-            WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+            WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
             Ext,
         ) -> eyre::Result<()>,
     {
@@ -77,13 +77,13 @@ where
     C: ChainSpecParser,
     Ext: clap::Args + fmt::Debug,
     F: AsyncFnOnce(
-        WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+        WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
         Ext,
     ) -> eyre::Result<()>,
 {
     fn entrypoint(
         self,
-        builder: WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+        builder: WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
         builder_args: Ext,
     ) -> impl Future<Output = eyre::Result<()>> {
         (self.func)(builder, builder_args)

crates/cli/commands/src/node.rs

@@ -206,7 +206,7 @@ where
         let db_path = data_dir.db();
 
         tracing::info!(target: "reth::cli", path = ?db_path, "Opening database");
-        let database = init_db(db_path.clone(), self.db.database_args())?.with_metrics();
+        let database = Arc::new(init_db(db_path.clone(), self.db.database_args())?.with_metrics());
 
         if with_unused_ports {
             node_config = node_config.with_unused_ports();

crates/cli/commands/src/prune.rs

@@ -1,62 +1,26 @@
-//! Command that runs pruning.
+//! Command that runs pruning without any limits.
 use crate::common::{AccessRights, CliNodeTypes, EnvironmentArgs};
 use clap::Parser;
-use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
+use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
-use reth_cli_runner::CliContext;
-use reth_node_builder::common::metrics_hooks;
-use reth_node_core::{args::MetricArgs, version::version_metadata};
-use reth_node_metrics::{
-    chain::ChainSpecInfo,
-    server::{MetricServer, MetricServerConfig},
-    version::VersionInfo,
-};
 use reth_prune::PrunerBuilder;
 use reth_static_file::StaticFileProducer;
 use std::sync::Arc;
 use tracing::info;
 
-/// Prunes according to the configuration
+/// Prunes according to the configuration without any limits
 #[derive(Debug, Parser)]
 pub struct PruneCommand<C: ChainSpecParser> {
     #[command(flatten)]
     env: EnvironmentArgs<C>,
-
-    /// Prometheus metrics configuration.
-    #[command(flatten)]
-    metrics: MetricArgs,
 }
 
 impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> PruneCommand<C> {
     /// Execute the `prune` command
-    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(
-        self,
-        ctx: CliContext,
-    ) -> eyre::Result<()> {
+    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(self) -> eyre::Result<()> {
         let env = self.env.init::<N>(AccessRights::RW)?;
         let provider_factory = env.provider_factory;
         let config = env.config.prune;
-        let data_dir = env.data_dir;
-
-        if let Some(listen_addr) = self.metrics.prometheus {
-            let config = MetricServerConfig::new(
-                listen_addr,
-                VersionInfo {
-                    version: version_metadata().cargo_pkg_version.as_ref(),
-                    build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
-                    cargo_features: version_metadata().vergen_cargo_features.as_ref(),
-                    git_sha: version_metadata().vergen_git_sha.as_ref(),
-                    target_triple: version_metadata().vergen_cargo_target_triple.as_ref(),
-                    build_profile: version_metadata().build_profile_name.as_ref(),
-                },
-                ChainSpecInfo { name: provider_factory.chain_spec().chain().to_string() },
-                ctx.task_executor,
-                metrics_hooks(&provider_factory),
-                data_dir.pprof_dumps(),
-            );
-
-            MetricServer::new(config).serve().await?;
-        }
 
         // Copy data from database to static files
         info!(target: "reth::cli", "Copying data from database to static files...");
@@ -69,43 +33,13 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> PruneComma
         // Delete data which has been copied to static files.
         if let Some(prune_tip) = lowest_static_file_height {
             info!(target: "reth::cli", ?prune_tip, ?config, "Pruning data from database...");
-
-            // Use batched pruning with a limit to bound memory, running in a loop until complete.
-            const DELETE_LIMIT: usize = 200_000;
+            // Run the pruner according to the configuration, and don't enforce any limits on it
             let mut pruner = PrunerBuilder::new(config)
-                .delete_limit(DELETE_LIMIT)
+                .delete_limit(usize::MAX)
                 .build_with_provider_factory(provider_factory);
 
-            let mut total_pruned = 0usize;
-            loop {
-                let output = pruner.run(prune_tip)?;
-                let batch_pruned: usize = output.segments.iter().map(|(_, seg)| seg.pruned).sum();
-                total_pruned = total_pruned.saturating_add(batch_pruned);
-
-                // Check if all segments are finished (not just the overall progress,
-                // since the pruner sets overall progress from the last segment only)
-                let all_segments_finished =
-                    output.segments.iter().all(|(_, seg)| seg.progress.is_finished());
-
-                if all_segments_finished {
-                    break;
-                }
-
-                if batch_pruned == 0 {
-                    return Err(eyre::eyre!(
-                        "pruner made no progress but reported more data remaining; \
-                         aborting to prevent infinite loop"
-                    ));
-                }
-
-                info!(
-                    target: "reth::cli",
-                    batch_pruned,
-                    total_pruned,
-                    "Pruning batch complete, continuing..."
-                );
-            }
-            info!(target: "reth::cli", total_pruned, "Pruned data from database");
+            pruner.run(prune_tip)?;
+            info!(target: "reth::cli", "Pruned data from database");
         }
 
         Ok(())

crates/cli/commands/src/stage/dump/execution.rs

@@ -26,7 +26,7 @@ pub(crate) async fn dump_execution_stage<N, E, C>(
     consensus: C,
 ) -> eyre::Result<()>
 where
-    N: ProviderNodeTypes<DB = DatabaseEnv>,
+    N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>,
     E: ConfigureEvm<Primitives = N::Primitives> + 'static,
     C: FullConsensus<E::Primitives> + 'static,
 {
@@ -39,7 +39,7 @@ where
     if should_run {
         dry_run(
             ProviderFactory::<N>::new(
-                output_db,
+                Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
                 RocksDBProvider::builder(output_datadir.rocksdb()).build()?,

crates/cli/commands/src/stage/dump/hashing_account.rs

@@ -10,9 +10,10 @@ use reth_provider::{
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::AccountHashingStage, Stage, StageCheckpoint, UnwindInput};
+use std::sync::Arc;
 use tracing::info;
 
-pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = DatabaseEnv>>(
+pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>>(
     db_tool: &DbTool<N>,
     from: BlockNumber,
     to: BlockNumber,
@@ -35,7 +36,7 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Databas
     if should_run {
         dry_run(
             ProviderFactory::<N>::new(
-                output_db,
+                Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
                 RocksDBProvider::builder(output_datadir.rocksdb()).build()?,

crates/cli/commands/src/stage/dump/hashing_storage.rs

@@ -9,9 +9,10 @@ use reth_provider::{
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::StorageHashingStage, Stage, StageCheckpoint, UnwindInput};
+use std::sync::Arc;
 use tracing::info;
 
-pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = DatabaseEnv>>(
+pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>>(
     db_tool: &DbTool<N>,
     from: u64,
     to: u64,
@@ -25,7 +26,7 @@ pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Databas
     if should_run {
         dry_run(
             ProviderFactory::<N>::new(
-                output_db,
+                Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
                 RocksDBProvider::builder(output_datadir.rocksdb()).build()?,

crates/cli/commands/src/stage/dump/merkle.rs

@@ -34,7 +34,7 @@ pub(crate) async fn dump_merkle_stage<N>(
     consensus: impl FullConsensus<N::Primitives> + 'static,
 ) -> Result<()>
 where
-    N: ProviderNodeTypes<DB = DatabaseEnv>,
+    N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>,
 {
     let (output_db, tip_block_number) = setup(from, to, &output_datadir.db(), db_tool)?;
 
@@ -59,7 +59,7 @@ where
     if should_run {
         dry_run(
             ProviderFactory::<N>::new(
-                output_db,
+                Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
                 RocksDBProvider::builder(output_datadir.rocksdb()).build()?,

crates/cli/commands/src/stage/unwind.rs

@@ -158,7 +158,7 @@ enum Subcommands {
 
 impl Subcommands {
     /// Returns the block to unwind to. The returned block will stay in database.
-    fn unwind_target<N: ProviderNodeTypes<DB = DatabaseEnv>>(
+    fn unwind_target<N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>>(
         &self,
         factory: ProviderFactory<N>,
     ) -> eyre::Result<u64> {

crates/consensus/debug-client/Cargo.toml

@@ -29,7 +29,7 @@ auto_impl.workspace = true
 derive_more.workspace = true
 futures.workspace = true
 eyre.workspace = true
-reqwest.workspace = true
+reqwest = { workspace = true, features = ["rustls-tls"] }
 serde = { workspace = true, features = ["derive"] }
 tokio = { workspace = true, features = ["time"] }
 serde_json.workspace = true

crates/consensus/debug-client/src/client.rs

@@ -38,8 +38,12 @@ pub trait BlockProvider: Send + Sync + 'static {
         offset: usize,
     ) -> impl Future<Output = eyre::Result<B256>> + Send {
         async move {
-            if let Some(hash) = get_hash_at_offset(previous_block_hashes, offset) {
-                return Ok(hash);
+            let stored_hash = previous_block_hashes
+                .len()
+                .checked_sub(offset)
+                .and_then(|index| previous_block_hashes.get(index));
+            if let Some(hash) = stored_hash {
+                return Ok(*hash);
             }
 
             // Return zero hash if the chain isn't long enough to have the block at the offset.
@@ -79,7 +83,7 @@ where
     /// Spawn the client to start sending FCUs and new payloads by periodically fetching recent
     /// blocks.
     pub async fn run(self) {
-        let mut previous_block_hashes = AllocRingBuffer::new(65);
+        let mut previous_block_hashes = AllocRingBuffer::new(64);
         let mut block_stream = {
             let (tx, rx) = mpsc::channel::<P::Block>(64);
             let block_provider = self.block_provider.clone();
@@ -95,7 +99,7 @@ where
             let block_hash = payload.block_hash();
             let block_number = payload.block_number();
 
-            previous_block_hashes.enqueue(block_hash);
+            previous_block_hashes.push(block_hash);
 
             // Send new events to execution client
             let _ = self.engine_handle.new_payload(payload).await;
@@ -138,60 +142,3 @@ where
         }
     }
 }
-
-/// Looks up a block hash from the ring buffer at the given offset from the most recent entry.
-///
-/// Returns `None` if the buffer doesn't have enough entries to satisfy the offset.
-fn get_hash_at_offset(buffer: &AllocRingBuffer<B256>, offset: usize) -> Option<B256> {
-    buffer.len().checked_sub(offset + 1).and_then(|index| buffer.get(index).copied())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_hash_at_offset() {
-        let mut buffer: AllocRingBuffer<B256> = AllocRingBuffer::new(65);
-
-        // Empty buffer returns None for any offset
-        assert_eq!(get_hash_at_offset(&buffer, 0), None);
-        assert_eq!(get_hash_at_offset(&buffer, 1), None);
-
-        // Push hashes 0..65
-        for i in 0..65u8 {
-            buffer.enqueue(B256::with_last_byte(i));
-        }
-
-        // offset=0 should return the most recent (64)
-        assert_eq!(get_hash_at_offset(&buffer, 0), Some(B256::with_last_byte(64)));
-
-        // offset=32 (safe block) should return hash 32
-        assert_eq!(get_hash_at_offset(&buffer, 32), Some(B256::with_last_byte(32)));
-
-        // offset=64 (finalized block) should return hash 0 (the oldest)
-        assert_eq!(get_hash_at_offset(&buffer, 64), Some(B256::with_last_byte(0)));
-
-        // offset=65 exceeds buffer, should return None
-        assert_eq!(get_hash_at_offset(&buffer, 65), None);
-    }
-
-    #[test]
-    fn test_get_hash_at_offset_insufficient_entries() {
-        let mut buffer: AllocRingBuffer<B256> = AllocRingBuffer::new(65);
-
-        // With only 1 entry, only offset=0 works
-        buffer.enqueue(B256::with_last_byte(1));
-        assert_eq!(get_hash_at_offset(&buffer, 0), Some(B256::with_last_byte(1)));
-        assert_eq!(get_hash_at_offset(&buffer, 1), None);
-        assert_eq!(get_hash_at_offset(&buffer, 32), None);
-        assert_eq!(get_hash_at_offset(&buffer, 64), None);
-
-        // With 33 entries, offset=32 works but offset=64 doesn't
-        for i in 2..=33u8 {
-            buffer.enqueue(B256::with_last_byte(i));
-        }
-        assert_eq!(get_hash_at_offset(&buffer, 32), Some(B256::with_last_byte(1)));
-        assert_eq!(get_hash_at_offset(&buffer, 64), None);
-    }
-}

crates/e2e-test-utils/src/setup_import.rs

@@ -114,22 +114,22 @@ pub async fn setup_engine_with_chain_import(
 
         // Initialize the database using init_db (same as CLI import command)
         let db_args = reth_node_core::args::DatabaseArgs::default().database_args();
-        let db = reth_db::init_db(&db_path, db_args)?;
+        let db_env = reth_db::init_db(&db_path, db_args)?;
+        let db = Arc::new(db_env);
 
         // Create a provider factory with the initialized database (use regular DB, not
         // TempDatabase) We need to specify the node types properly for the adapter
-        let provider_factory =
-            ProviderFactory::<NodeTypesWithDBAdapter<EthereumNode, DatabaseEnv>>::new(
-                db.clone(),
-                chain_spec.clone(),
-                reth_provider::providers::StaticFileProvider::read_write(
-                    static_files_path.clone(),
-                )?,
-                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
-                    .with_default_tables()
-                    .build()
-                    .unwrap(),
-            )?;
+        let provider_factory = ProviderFactory::<
+            NodeTypesWithDBAdapter<EthereumNode, Arc<DatabaseEnv>>,
+        >::new(
+            db.clone(),
+            chain_spec.clone(),
+            reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())?,
+            reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
+                .with_default_tables()
+                .build()
+                .unwrap(),
+        )?;
 
         // Initialize genesis if needed
         reth_db_common::init::init_genesis(&provider_factory)?;
@@ -320,10 +320,11 @@ mod tests {
         // Import the chain
         {
             let db_args = reth_node_core::args::DatabaseArgs::default().database_args();
-            let db = reth_db::init_db(&db_path, db_args).unwrap();
+            let db_env = reth_db::init_db(&db_path, db_args).unwrap();
+            let db = Arc::new(db_env);
 
             let provider_factory: ProviderFactory<
-                NodeTypesWithDBAdapter<reth_node_ethereum::EthereumNode, DatabaseEnv>,
+                NodeTypesWithDBAdapter<reth_node_ethereum::EthereumNode, Arc<DatabaseEnv>>,
             > = ProviderFactory::new(
                 db.clone(),
                 chain_spec.clone(),
@@ -384,10 +385,11 @@ mod tests {
 
         // Now reopen the database and verify checkpoints are still there
         {
-            let db = reth_db::init_db(&db_path, DatabaseArguments::default()).unwrap();
+            let db_env = reth_db::init_db(&db_path, DatabaseArguments::default()).unwrap();
+            let db = Arc::new(db_env);
 
             let provider_factory: ProviderFactory<
-                NodeTypesWithDBAdapter<reth_node_ethereum::EthereumNode, DatabaseEnv>,
+                NodeTypesWithDBAdapter<reth_node_ethereum::EthereumNode, Arc<DatabaseEnv>>,
             > = ProviderFactory::new(
                 db,
                 chain_spec.clone(),

crates/e2e-test-utils/tests/rocksdb/main.rs

@@ -469,123 +469,3 @@ async fn test_rocksdb_pending_tx_not_in_storage() -> Result<()> {
 
     Ok(())
 }
-
-/// Reorg with `RocksDB`: verifies that unwind correctly reads changesets from
-/// storage-aware locations (static files vs MDBX) rather than directly from MDBX.
-///
-/// This test exercises `unwind_trie_state_from` which previously failed with
-/// `UnsortedInput` errors because it read changesets directly from MDBX tables
-/// instead of using storage-aware methods that check `storage_changesets_in_static_files`.
-#[tokio::test]
-async fn test_rocksdb_reorg_unwind() -> Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let chain_spec = test_chain_spec();
-    let chain_id = chain_spec.chain().id();
-
-    let (mut nodes, _tasks, _) = E2ETestSetupBuilder::<EthereumNode, _>::new(
-        1,
-        chain_spec.clone(),
-        test_attributes_generator,
-    )
-    .with_tree_config_modifier(|config| config.with_persistence_threshold(0))
-    .build()
-    .await?;
-
-    assert_eq!(nodes.len(), 1);
-
-    // Use two separate wallets to avoid nonce conflicts during reorg
-    let wallets = wallet::Wallet::new(2).with_chain_id(chain_id).wallet_gen();
-    let signer1 = wallets[0].clone();
-    let signer2 = wallets[1].clone();
-    let client = nodes[0].rpc_client().expect("RPC client");
-
-    // Mine block 1 with a transaction from signer1
-    let raw_tx1 =
-        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 0).await;
-    let tx_hash1 = nodes[0].rpc.inject_tx(raw_tx1).await?;
-    wait_for_pending_tx(&client, tx_hash1).await;
-
-    let payload1 = nodes[0].advance_block().await?;
-    let block1_hash = payload1.block().hash();
-    assert_eq!(payload1.block().number(), 1);
-
-    // Poll until tx1 appears in RocksDB (ensures persistence happened)
-    let tx_number1 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash1).await;
-    assert_eq!(tx_number1, 0, "First tx should have tx_number 0");
-
-    // Mine block 2 with transaction from signer1 (nonce 1)
-    let raw_tx2 =
-        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 1).await;
-    let tx_hash2 = nodes[0].rpc.inject_tx(raw_tx2).await?;
-    wait_for_pending_tx(&client, tx_hash2).await;
-
-    let payload2 = nodes[0].advance_block().await?;
-    assert_eq!(payload2.block().number(), 2);
-
-    // Poll until tx2 appears in RocksDB
-    let tx_number2 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash2).await;
-    assert_eq!(tx_number2, 1, "Second tx should have tx_number 1");
-
-    // Mine block 3 with transaction from signer1 (nonce 2)
-    let raw_tx3 =
-        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 2).await;
-    let tx_hash3 = nodes[0].rpc.inject_tx(raw_tx3).await?;
-    wait_for_pending_tx(&client, tx_hash3).await;
-
-    let payload3 = nodes[0].advance_block().await?;
-    assert_eq!(payload3.block().number(), 3);
-
-    // Poll until tx3 appears in RocksDB
-    let tx_number3 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash3).await;
-    assert_eq!(tx_number3, 2, "Third tx should have tx_number 2");
-
-    // Now create an alternate block 2 using signer2 (different wallet, avoids nonce conflict)
-    // Inject a tx from signer2 (nonce 0) before building the alternate block
-    let raw_alt_tx =
-        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer2.clone(), 0).await;
-    let alt_tx_hash = nodes[0].rpc.inject_tx(raw_alt_tx).await?;
-    wait_for_pending_tx(&client, alt_tx_hash).await;
-
-    // Build an alternate payload (this builds on top of the current head, i.e., block 3)
-    // But we want to reorg back to block 1, so we'll use the payload and then FCU to it
-    let alt_payload = nodes[0].new_payload().await?;
-    let alt_block_hash = nodes[0].submit_payload(alt_payload.clone()).await?;
-
-    // Trigger reorg: make the alternate chain canonical by sending FCU pointing to block 1's hash
-    // as finalized, which should trigger an unwind of blocks 2 and 3
-    // The alt block becomes the new head
-    nodes[0].update_forkchoice(block1_hash, alt_block_hash).await?;
-
-    // Give time for the reorg to complete
-    tokio::time::sleep(Duration::from_millis(500)).await;
-
-    // Verify we can still query transactions and the chain is consistent
-    // If unwind_trie_state_from failed, this would have errored during reorg
-    let latest: Option<alloy_rpc_types_eth::Block> =
-        client.request("eth_getBlockByNumber", ("latest", false)).await?;
-    let latest = latest.expect("Latest block should exist");
-    // The alt block is at height 4 (on top of block 3)
-    assert!(latest.header.number >= 3, "Should be at height >= 3 after operation");
-
-    // tx1 from block 1 should still be there
-    let tx1: Option<Transaction> = client.request("eth_getTransactionByHash", [tx_hash1]).await?;
-    assert!(tx1.is_some(), "tx1 from block 1 should still be queryable");
-    assert_eq!(tx1.unwrap().block_number, Some(1));
-
-    // Mine another block to verify the chain can continue
-    let raw_tx_final =
-        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer2.clone(), 1).await;
-    let tx_hash_final = nodes[0].rpc.inject_tx(raw_tx_final).await?;
-    wait_for_pending_tx(&client, tx_hash_final).await;
-
-    let final_payload = nodes[0].advance_block().await?;
-    assert!(final_payload.block().number() > 3, "Should be able to mine block after reorg");
-
-    // Verify tx_final is included
-    let tx_final: Option<Transaction> =
-        client.request("eth_getTransactionByHash", [tx_hash_final]).await?;
-    assert!(tx_final.is_some(), "final tx should be in latest block");
-
-    Ok(())
-}

crates/engine/primitives/src/config.rs

@@ -47,17 +47,6 @@ pub const DEFAULT_RESERVED_CPU_CORES: usize = 1;
 /// Default maximum concurrency for prewarm task.
 pub const DEFAULT_PREWARM_MAX_CONCURRENCY: usize = 16;
 
-/// Default depth for sparse trie pruning.
-///
-/// Nodes at this depth and below are converted to hash stubs to reduce memory.
-/// Depth 4 means we keep roughly 16^4 = 65536 potential branch paths at most.
-pub const DEFAULT_SPARSE_TRIE_PRUNE_DEPTH: usize = 4;
-
-/// Default maximum number of storage tries to keep after pruning.
-///
-/// Storage tries beyond this limit are cleared (but allocations preserved).
-pub const DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES: usize = 100;
-
 const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = EPOCH_SLOTS as u32 * 2;
 const DEFAULT_MAX_INVALID_HEADER_CACHE_LENGTH: u32 = 256;
 const DEFAULT_MAX_EXECUTE_BLOCK_BATCH_SIZE: usize = 4;
@@ -165,10 +154,6 @@ pub struct TreeConfig {
     disable_cache_metrics: bool,
     /// Whether to enable sparse trie as cache.
     enable_sparse_trie_as_cache: bool,
-    /// Depth for sparse trie pruning after state root computation.
-    sparse_trie_prune_depth: usize,
-    /// Maximum number of storage tries to retain after pruning.
-    sparse_trie_max_storage_tries: usize,
 }
 
 impl Default for TreeConfig {
@@ -198,9 +183,7 @@ impl Default for TreeConfig {
             account_worker_count: default_account_worker_count(),
             disable_proof_v2: false,
             disable_cache_metrics: false,
-            enable_sparse_trie_as_cache: false,
-            sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
-            sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
+            enable_sparse_trie_as_cache: true,
         }
     }
 }
@@ -233,8 +216,6 @@ impl TreeConfig {
         account_worker_count: usize,
         disable_proof_v2: bool,
         disable_cache_metrics: bool,
-        sparse_trie_prune_depth: usize,
-        sparse_trie_max_storage_tries: usize,
     ) -> Self {
         Self {
             persistence_threshold,
@@ -262,8 +243,6 @@ impl TreeConfig {
             disable_proof_v2,
             disable_cache_metrics,
             enable_sparse_trie_as_cache: false,
-            sparse_trie_prune_depth,
-            sparse_trie_max_storage_tries,
         }
     }
 
@@ -570,32 +549,4 @@ impl TreeConfig {
     pub const fn enable_sparse_trie_as_cache(&self) -> bool {
         self.enable_sparse_trie_as_cache
     }
-
-    /// Setter for whether to enable sparse trie as cache.
-    pub const fn with_enable_sparse_trie_as_cache(mut self, value: bool) -> Self {
-        self.enable_sparse_trie_as_cache = value;
-        self
-    }
-
-    /// Returns the sparse trie prune depth.
-    pub const fn sparse_trie_prune_depth(&self) -> usize {
-        self.sparse_trie_prune_depth
-    }
-
-    /// Setter for sparse trie prune depth.
-    pub const fn with_sparse_trie_prune_depth(mut self, depth: usize) -> Self {
-        self.sparse_trie_prune_depth = depth;
-        self
-    }
-
-    /// Returns the maximum number of storage tries to retain after pruning.
-    pub const fn sparse_trie_max_storage_tries(&self) -> usize {
-        self.sparse_trie_max_storage_tries
-    }
-
-    /// Setter for maximum storage tries to retain.
-    pub const fn with_sparse_trie_max_storage_tries(mut self, max_tries: usize) -> Self {
-        self.sparse_trie_max_storage_tries = max_tries;
-        self
-    }
 }

crates/engine/service/Cargo.toml

@@ -17,6 +17,7 @@ reth-engine-tree.workspace = true
 reth-evm.workspace = true
 reth-network-p2p.workspace = true
 reth-payload-builder.workspace = true
+reth-ethereum-primitives.workspace = true
 reth-provider.workspace = true
 reth-prune.workspace = true
 reth-stages-api.workspace = true

crates/engine/service/src/service.rs

@@ -14,6 +14,7 @@ pub use reth_engine_tree::{
     chain::{ChainEvent, ChainOrchestrator},
     engine::EngineApiEvent,
 };
+use reth_ethereum_primitives::EthPrimitives;
 use reth_evm::ConfigureEvm;
 use reth_network_p2p::BlockClient;
 use reth_node_types::{BlockTy, NodeTypes};
@@ -96,7 +97,7 @@ where
         let downloader = BasicBlockDownloader::new(client, consensus.clone());
 
         let persistence_handle =
-            PersistenceHandle::<N::Primitives>::spawn_service(provider, pruner, sync_metrics_tx);
+            PersistenceHandle::<EthPrimitives>::spawn_service(provider, pruner, sync_metrics_tx);
 
         let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();
 

crates/engine/tree/Cargo.toml

@@ -23,7 +23,7 @@ reth-evm = { workspace = true, features = ["metrics"] }
 reth-network-p2p.workspace = true
 reth-payload-builder.workspace = true
 reth-payload-primitives.workspace = true
-reth-primitives-traits = { workspace = true, features = ["rayon"] }
+reth-primitives-traits.workspace = true
 reth-ethereum-primitives.workspace = true
 reth-provider.workspace = true
 reth-prune.workspace = true

crates/engine/tree/src/metrics.rs

@@ -20,7 +20,7 @@ pub(crate) struct PersistenceMetrics {
     /// How long it took for blocks to be saved
     pub(crate) save_blocks_duration_seconds: Histogram,
     /// How many blocks we persist at once.
-    pub(crate) save_blocks_batch_size: Histogram,
+    pub(crate) save_blocks_block_count: Histogram,
     /// How long it took for blocks to be pruned
     pub(crate) prune_before_duration_seconds: Histogram,
 }

crates/engine/tree/src/persistence.rs

@@ -12,11 +12,7 @@ use reth_provider::{
 use reth_prune::{PrunerError, PrunerOutput, PrunerWithFactory};
 use reth_stages_api::{MetricEvent, MetricEventsSender};
 use std::{
-    sync::{
-        mpsc::{Receiver, SendError, Sender},
-        Arc,
-    },
-    thread::JoinHandle,
+    sync::mpsc::{Receiver, SendError, Sender},
     time::Instant,
 };
 use thiserror::Error;
@@ -44,12 +40,6 @@ where
     metrics: PersistenceMetrics,
     /// Sender for sync metrics - we only submit sync metrics for persisted blocks
     sync_metrics_tx: MetricEventsSender,
-    /// Pending finalized block number to be committed with the next block save.
-    /// This avoids triggering a separate fsync for each finalized block update.
-    pending_finalized_block: Option<u64>,
-    /// Pending safe block number to be committed with the next block save.
-    /// This avoids triggering a separate fsync for each safe block update.
-    pending_safe_block: Option<u64>,
 }
 
 impl<N> PersistenceService<N>
@@ -63,15 +53,7 @@ where
         pruner: PrunerWithFactory<ProviderFactory<N>>,
         sync_metrics_tx: MetricEventsSender,
     ) -> Self {
-        Self {
-            provider,
-            incoming,
-            pruner,
-            metrics: PersistenceMetrics::default(),
-            sync_metrics_tx,
-            pending_finalized_block: None,
-            pending_safe_block: None,
-        }
+        Self { provider, incoming, pruner, metrics: PersistenceMetrics::default(), sync_metrics_tx }
     }
 
     /// Prunes block data before the given block number according to the configured prune
@@ -124,10 +106,14 @@ where
                     }
                 }
                 PersistenceAction::SaveFinalizedBlock(finalized_block) => {
-                    self.pending_finalized_block = Some(finalized_block);
+                    let provider = self.provider.database_provider_rw()?;
+                    provider.save_finalized_block_number(finalized_block)?;
+                    provider.commit()?;
                 }
                 PersistenceAction::SaveSafeBlock(safe_block) => {
-                    self.pending_safe_block = Some(safe_block);
+                    let provider = self.provider.database_provider_rw()?;
+                    provider.save_safe_block_number(safe_block)?;
+                    provider.commit()?;
                 }
             }
         }
@@ -152,39 +138,26 @@ where
     }
 
     fn on_save_blocks(
-        &mut self,
+        &self,
         blocks: Vec<ExecutedBlock<N::Primitives>>,
     ) -> Result<Option<BlockNumHash>, PersistenceError> {
         let first_block = blocks.first().map(|b| b.recovered_block.num_hash());
         let last_block = blocks.last().map(|b| b.recovered_block.num_hash());
         let block_count = blocks.len();
-
-        // Take any pending finalized/safe block updates to commit together
-        let pending_finalized = self.pending_finalized_block.take();
-        let pending_safe = self.pending_safe_block.take();
-
         debug!(target: "engine::persistence", ?block_count, first=?first_block, last=?last_block, "Saving range of blocks");
 
         let start_time = Instant::now();
 
         if last_block.is_some() {
             let provider_rw = self.provider.database_provider_rw()?;
+
             provider_rw.save_blocks(blocks, SaveBlocksMode::Full)?;
-
-            // Commit pending finalized/safe block updates in the same transaction
-            if let Some(finalized) = pending_finalized {
-                provider_rw.save_finalized_block_number(finalized)?;
-            }
-            if let Some(safe) = pending_safe {
-                provider_rw.save_safe_block_number(safe)?;
-            }
-
             provider_rw.commit()?;
         }
 
         debug!(target: "engine::persistence", first=?first_block, last=?last_block, "Saved range of blocks");
 
-        self.metrics.save_blocks_batch_size.record(block_count as f64);
+        self.metrics.save_blocks_block_count.record(block_count as f64);
         self.metrics.save_blocks_duration_seconds.record(start_time.elapsed());
 
         Ok(last_block)
@@ -231,25 +204,15 @@ pub enum PersistenceAction<N: NodePrimitives = EthPrimitives> {
 pub struct PersistenceHandle<N: NodePrimitives = EthPrimitives> {
     /// The channel used to communicate with the persistence service
     sender: Sender<PersistenceAction<N>>,
-    /// Guard that joins the service thread when all handles are dropped.
-    /// Uses `Arc` so the handle remains `Clone`.
-    _service_guard: Arc<ServiceGuard>,
 }
 
 impl<T: NodePrimitives> PersistenceHandle<T> {
     /// Create a new [`PersistenceHandle`] from a [`Sender<PersistenceAction>`].
-    ///
-    /// This is intended for testing purposes where you want to mock the persistence service.
-    /// For production use, prefer [`spawn_service`](Self::spawn_service).
-    pub fn new(sender: Sender<PersistenceAction<T>>) -> Self {
-        Self { sender, _service_guard: Arc::new(ServiceGuard(None)) }
+    pub const fn new(sender: Sender<PersistenceAction<T>>) -> Self {
+        Self { sender }
     }
 
     /// Create a new [`PersistenceHandle`], and spawn the persistence service.
-    ///
-    /// The returned handle can be cloned and shared. When all clones are dropped, the service
-    /// thread will be joined, ensuring graceful shutdown before resources (like `RocksDB`) are
-    /// released.
     pub fn spawn_service<N>(
         provider_factory: ProviderFactory<N>,
         pruner: PrunerWithFactory<ProviderFactory<N>>,
@@ -261,10 +224,13 @@ impl<T: NodePrimitives> PersistenceHandle<T> {
         // create the initial channels
         let (db_service_tx, db_service_rx) = std::sync::mpsc::channel();
 
+        // construct persistence handle
+        let persistence_handle = PersistenceHandle::new(db_service_tx);
+
         // spawn the persistence service
         let db_service =
             PersistenceService::new(provider_factory, db_service_rx, pruner, sync_metrics_tx);
-        let join_handle = std::thread::Builder::new()
+        std::thread::Builder::new()
             .name("Persistence Service".to_string())
             .spawn(|| {
                 if let Err(err) = db_service.run() {
@@ -273,10 +239,7 @@ impl<T: NodePrimitives> PersistenceHandle<T> {
             })
             .unwrap();
 
-        PersistenceHandle {
-            sender: db_service_tx,
-            _service_guard: Arc::new(ServiceGuard(Some(join_handle))),
-        }
+        persistence_handle
     }
 
     /// Sends a specific [`PersistenceAction`] in the contained channel. The caller is responsible
@@ -304,10 +267,7 @@ impl<T: NodePrimitives> PersistenceHandle<T> {
         self.send_action(PersistenceAction::SaveBlocks(blocks, tx))
     }
 
-    /// Queues the finalized block number to be persisted on disk.
-    ///
-    /// The update is deferred and will be committed together with the next [`Self::save_blocks`]
-    /// call to avoid triggering a separate fsync for each update.
+    /// Persists the finalized block number on disk.
     pub fn save_finalized_block_number(
         &self,
         finalized_block: u64,
@@ -315,10 +275,7 @@ impl<T: NodePrimitives> PersistenceHandle<T> {
         self.send_action(PersistenceAction::SaveFinalizedBlock(finalized_block))
     }
 
-    /// Queues the safe block number to be persisted on disk.
-    ///
-    /// The update is deferred and will be committed together with the next [`Self::save_blocks`]
-    /// call to avoid triggering a separate fsync for each update.
+    /// Persists the safe block number on disk.
     pub fn save_safe_block_number(
         &self,
         safe_block: u64,
@@ -340,27 +297,6 @@ impl<T: NodePrimitives> PersistenceHandle<T> {
     }
 }
 
-/// Guard that joins the persistence service thread when dropped.
-///
-/// This ensures graceful shutdown - the service thread completes before resources like
-/// `RocksDB` are released. Stored in an `Arc` inside [`PersistenceHandle`] so the handle
-/// can be cloned while sharing the same guard.
-struct ServiceGuard(Option<JoinHandle<()>>);
-
-impl std::fmt::Debug for ServiceGuard {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_tuple("ServiceGuard").field(&self.0.as_ref().map(|_| "...")).finish()
-    }
-}
-
-impl Drop for ServiceGuard {
-    fn drop(&mut self) {
-        if let Some(join_handle) = self.0.take() {
-            let _ = join_handle.join();
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -387,12 +323,12 @@ mod tests {
     #[test]
     fn test_save_blocks_empty() {
         reth_tracing::init_test_tracing();
-        let handle = default_persistence_handle();
+        let persistence_handle = default_persistence_handle();
 
         let blocks = vec![];
         let (tx, rx) = crossbeam_channel::bounded(1);
 
-        handle.save_blocks(blocks, tx).unwrap();
+        persistence_handle.save_blocks(blocks, tx).unwrap();
 
         let hash = rx.recv().unwrap();
         assert_eq!(hash, None);
@@ -401,7 +337,7 @@ mod tests {
     #[test]
     fn test_save_blocks_single_block() {
         reth_tracing::init_test_tracing();
-        let handle = default_persistence_handle();
+        let persistence_handle = default_persistence_handle();
         let block_number = 0;
         let mut test_block_builder = TestBlockBuilder::eth();
         let executed =
@@ -411,7 +347,7 @@ mod tests {
         let blocks = vec![executed];
         let (tx, rx) = crossbeam_channel::bounded(1);
 
-        handle.save_blocks(blocks, tx).unwrap();
+        persistence_handle.save_blocks(blocks, tx).unwrap();
 
         let BlockNumHash { hash: actual_hash, number: _ } = rx
             .recv_timeout(std::time::Duration::from_secs(10))
@@ -424,14 +360,14 @@ mod tests {
     #[test]
     fn test_save_blocks_multiple_blocks() {
         reth_tracing::init_test_tracing();
-        let handle = default_persistence_handle();
+        let persistence_handle = default_persistence_handle();
 
         let mut test_block_builder = TestBlockBuilder::eth();
         let blocks = test_block_builder.get_executed_blocks(0..5).collect::<Vec<_>>();
         let last_hash = blocks.last().unwrap().recovered_block().hash();
         let (tx, rx) = crossbeam_channel::bounded(1);
 
-        handle.save_blocks(blocks, tx).unwrap();
+        persistence_handle.save_blocks(blocks, tx).unwrap();
         let BlockNumHash { hash: actual_hash, number: _ } = rx.recv().unwrap().unwrap();
         assert_eq!(last_hash, actual_hash);
     }
@@ -439,7 +375,7 @@ mod tests {
     #[test]
     fn test_save_blocks_multiple_calls() {
         reth_tracing::init_test_tracing();
-        let handle = default_persistence_handle();
+        let persistence_handle = default_persistence_handle();
 
         let ranges = [0..1, 1..2, 2..4, 4..5];
         let mut test_block_builder = TestBlockBuilder::eth();
@@ -448,7 +384,7 @@ mod tests {
             let last_hash = blocks.last().unwrap().recovered_block().hash();
             let (tx, rx) = crossbeam_channel::bounded(1);
 
-            handle.save_blocks(blocks, tx).unwrap();
+            persistence_handle.save_blocks(blocks, tx).unwrap();
 
             let BlockNumHash { hash: actual_hash, number: _ } = rx.recv().unwrap().unwrap();
             assert_eq!(last_hash, actual_hash);

crates/engine/tree/src/tree/cached_state.rs

@@ -76,8 +76,7 @@ impl CacheConfig for EpochCacheConfig {
 type FixedCache<K, V, H = DefaultHashBuilder> = fixed_cache::Cache<K, V, H, EpochCacheConfig>;
 
 /// A wrapper of a state provider and a shared cache.
-#[derive(Debug)]
-pub struct CachedStateProvider<S> {
+pub(crate) struct CachedStateProvider<S> {
     /// The state provider
     state_provider: S,
 
@@ -97,7 +96,7 @@ where
 {
     /// Creates a new [`CachedStateProvider`] from an [`ExecutionCache`], state provider, and
     /// [`CachedStateMetrics`].
-    pub const fn new(
+    pub(crate) const fn new(
         state_provider: S,
         caches: ExecutionCache,
         metrics: CachedStateMetrics,
@@ -115,7 +114,7 @@ impl<S> CachedStateProvider<S> {
     /// [`State`](revm::database::State) also caches internally during block execution and the cache
     /// is then updated after the block with the entire [`BundleState`] output of that block which
     /// contains all accessed accounts,code,storage. See also [`ExecutionCache::insert_state`].
-    pub const fn prewarm(mut self) -> Self {
+    pub(crate) const fn prewarm(mut self) -> Self {
         self.prewarm = true;
         self
     }
@@ -132,7 +131,7 @@ impl<S> CachedStateProvider<S> {
 /// and the fixed-cache internal stats (collisions, size, capacity).
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.caching")]
-pub struct CachedStateMetrics {
+pub(crate) struct CachedStateMetrics {
     /// Number of times a new execution cache was created
     execution_cache_created_total: Counter,
 
@@ -187,7 +186,7 @@ pub struct CachedStateMetrics {
 
 impl CachedStateMetrics {
     /// Sets all values to zero, indicating that a new block is being executed.
-    pub fn reset(&self) {
+    pub(crate) fn reset(&self) {
         // code cache
         self.code_cache_hits.set(0);
         self.code_cache_misses.set(0);
@@ -205,7 +204,7 @@ impl CachedStateMetrics {
     }
 
     /// Returns a new zeroed-out instance of [`CachedStateMetrics`].
-    pub fn zeroed() -> Self {
+    pub(crate) fn zeroed() -> Self {
         let zeroed = Self::default();
         zeroed.reset();
         zeroed
@@ -327,7 +326,7 @@ impl<S: AccountReader> AccountReader for CachedStateProvider<S> {
 
 /// Represents the status of a key in the cache.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub enum CachedStatus<T> {
+pub(crate) enum CachedStatus<T> {
     /// The key is not in the cache (or was invalidated). The value was recalculated.
     NotCached(T),
     /// The key exists in cache and has a specific value.
@@ -488,7 +487,7 @@ impl<S: HashedPostStateProvider> HashedPostStateProvider for CachedStateProvider
 /// Since EIP-6780, SELFDESTRUCT only works within the same transaction where the
 /// contract was created, so we don't need to handle clearing the storage.
 #[derive(Debug, Clone)]
-pub struct ExecutionCache {
+pub(crate) struct ExecutionCache {
     /// Cache for contract bytecode, keyed by code hash.
     code_cache: Arc<FixedCache<B256, Option<Bytecode>, FbBuildHasher<32>>>,
 
@@ -520,7 +519,7 @@ impl ExecutionCache {
     ///
     /// Fixed-cache requires power-of-two sizes for efficient indexing.
     /// With epochs enabled, the minimum size is 4096 entries.
-    pub const fn bytes_to_entries(size_bytes: usize, entry_size: usize) -> usize {
+    pub(crate) const fn bytes_to_entries(size_bytes: usize, entry_size: usize) -> usize {
         let entries = size_bytes / entry_size;
         // Round down to nearest power of two
         let rounded = if entries == 0 { 1 } else { (entries + 1).next_power_of_two() >> 1 };
@@ -533,7 +532,7 @@ impl ExecutionCache {
     }
 
     /// Build an [`ExecutionCache`] struct, so that execution caches can be easily cloned.
-    pub fn new(total_cache_size: usize) -> Self {
+    pub(crate) fn new(total_cache_size: usize) -> Self {
         let storage_cache_size = (total_cache_size * 8888) / 10000; // 88.88% of total
         let account_cache_size = (total_cache_size * 556) / 10000; // 5.56% of total
         let code_cache_size = (total_cache_size * 556) / 10000; // 5.56% of total
@@ -567,7 +566,7 @@ impl ExecutionCache {
     }
 
     /// Gets code from cache, or inserts using the provided function.
-    pub fn get_or_try_insert_code_with<E>(
+    pub(crate) fn get_or_try_insert_code_with<E>(
         &self,
         hash: B256,
         f: impl FnOnce() -> Result<Option<Bytecode>, E>,
@@ -586,7 +585,7 @@ impl ExecutionCache {
     }
 
     /// Gets storage from cache, or inserts using the provided function.
-    pub fn get_or_try_insert_storage_with<E>(
+    pub(crate) fn get_or_try_insert_storage_with<E>(
         &self,
         address: Address,
         key: StorageKey,
@@ -606,7 +605,7 @@ impl ExecutionCache {
     }
 
     /// Gets account from cache, or inserts using the provided function.
-    pub fn get_or_try_insert_account_with<E>(
+    pub(crate) fn get_or_try_insert_account_with<E>(
         &self,
         address: Address,
         f: impl FnOnce() -> Result<Option<Account>, E>,
@@ -625,7 +624,12 @@ impl ExecutionCache {
     }
 
     /// Insert storage value into cache.
-    pub fn insert_storage(&self, address: Address, key: StorageKey, value: Option<StorageValue>) {
+    pub(crate) fn insert_storage(
+        &self,
+        address: Address,
+        key: StorageKey,
+        value: Option<StorageValue>,
+    ) {
         self.storage_cache.insert((address, key), value.unwrap_or_default());
     }
 
@@ -658,8 +662,7 @@ impl ExecutionCache {
     ///
     /// Returns an error if the state updates are inconsistent and should be discarded.
     #[instrument(level = "debug", target = "engine::caching", skip_all)]
-    #[expect(clippy::result_unit_err)]
-    pub fn insert_state(&self, state_updates: &BundleState) -> Result<(), ()> {
+    pub(crate) fn insert_state(&self, state_updates: &BundleState) -> Result<(), ()> {
         let _enter =
             debug_span!(target: "engine::tree", "contracts", len = state_updates.contracts.len())
                 .entered();
@@ -768,7 +771,7 @@ impl ExecutionCache {
 /// A saved cache that has been used for executing a specific block, which has been updated for its
 /// execution.
 #[derive(Debug, Clone)]
-pub struct SavedCache {
+pub(crate) struct SavedCache {
     /// The hash of the block these caches were used to execute.
     hash: B256,
 
@@ -788,43 +791,43 @@ pub struct SavedCache {
 
 impl SavedCache {
     /// Creates a new instance with the internals
-    pub fn new(hash: B256, caches: ExecutionCache, metrics: CachedStateMetrics) -> Self {
+    pub(super) fn new(hash: B256, caches: ExecutionCache, metrics: CachedStateMetrics) -> Self {
         Self { hash, caches, metrics, usage_guard: Arc::new(()), disable_cache_metrics: false }
     }
 
     /// Sets whether to disable cache metrics recording.
-    pub const fn with_disable_cache_metrics(mut self, disable: bool) -> Self {
+    pub(super) const fn with_disable_cache_metrics(mut self, disable: bool) -> Self {
         self.disable_cache_metrics = disable;
         self
     }
 
     /// Returns the hash for this cache
-    pub const fn executed_block_hash(&self) -> B256 {
+    pub(crate) const fn executed_block_hash(&self) -> B256 {
         self.hash
     }
 
     /// Splits the cache into its caches, metrics, and `disable_cache_metrics` flag, consuming it.
-    pub fn split(self) -> (ExecutionCache, CachedStateMetrics, bool) {
+    pub(crate) fn split(self) -> (ExecutionCache, CachedStateMetrics, bool) {
         (self.caches, self.metrics, self.disable_cache_metrics)
     }
 
     /// Returns true if the cache is available for use (no other tasks are currently using it).
-    pub fn is_available(&self) -> bool {
+    pub(crate) fn is_available(&self) -> bool {
         Arc::strong_count(&self.usage_guard) == 1
     }
 
     /// Returns the current strong count of the usage guard.
-    pub fn usage_count(&self) -> usize {
+    pub(crate) fn usage_count(&self) -> usize {
         Arc::strong_count(&self.usage_guard)
     }
 
     /// Returns the [`ExecutionCache`] belonging to the tracked hash.
-    pub const fn cache(&self) -> &ExecutionCache {
+    pub(crate) const fn cache(&self) -> &ExecutionCache {
         &self.caches
     }
 
     /// Returns the metrics associated with this cache.
-    pub const fn metrics(&self) -> &CachedStateMetrics {
+    pub(crate) const fn metrics(&self) -> &CachedStateMetrics {
         &self.metrics
     }
 

crates/engine/tree/src/tree/metrics.rs

@@ -13,13 +13,13 @@ use std::time::{Duration, Instant};
 
 /// Metrics for the `EngineApi`.
 #[derive(Debug, Default)]
-pub struct EngineApiMetrics {
+pub(crate) struct EngineApiMetrics {
     /// Engine API-specific metrics.
-    pub engine: EngineMetrics,
+    pub(crate) engine: EngineMetrics,
     /// Block executor metrics.
-    pub executor: ExecutorMetrics,
+    pub(crate) executor: ExecutorMetrics,
     /// Metrics for block validation
-    pub block_validation: BlockValidationMetrics,
+    pub(crate) block_validation: BlockValidationMetrics,
     /// Canonical chain and reorg related metrics
     pub tree: TreeMetrics,
     /// Metrics for EIP-7928 Block-Level Access Lists (BAL).
@@ -32,7 +32,7 @@ impl EngineApiMetrics {
     ///
     /// This method updates metrics for execution time, gas usage, and the number
     /// of accounts, storage slots and bytecodes updated.
-    pub fn record_block_execution<R>(
+    pub(crate) fn record_block_execution<R>(
         &self,
         output: &BlockExecutionOutput<R>,
         execution_duration: Duration,
@@ -59,27 +59,27 @@ impl EngineApiMetrics {
     }
 
     /// Returns a reference to the executor metrics for use in state hooks.
-    pub const fn executor_metrics(&self) -> &ExecutorMetrics {
+    pub(crate) const fn executor_metrics(&self) -> &ExecutorMetrics {
         &self.executor
     }
 
     /// Records the duration of block pre-execution changes (e.g., beacon root update).
-    pub fn record_pre_execution(&self, elapsed: Duration) {
+    pub(crate) fn record_pre_execution(&self, elapsed: Duration) {
         self.executor.pre_execution_histogram.record(elapsed);
     }
 
     /// Records the duration of block post-execution changes (e.g., finalization).
-    pub fn record_post_execution(&self, elapsed: Duration) {
+    pub(crate) fn record_post_execution(&self, elapsed: Duration) {
         self.executor.post_execution_histogram.record(elapsed);
     }
 
     /// Records the time spent waiting for the next transaction from the iterator.
-    pub fn record_transaction_wait(&self, elapsed: Duration) {
+    pub(crate) fn record_transaction_wait(&self, elapsed: Duration) {
         self.executor.transaction_wait_histogram.record(elapsed);
     }
 
     /// Records the duration of a single transaction execution.
-    pub fn record_transaction_execution(&self, elapsed: Duration) {
+    pub(crate) fn record_transaction_execution(&self, elapsed: Duration) {
         self.executor.transaction_execution_histogram.record(elapsed);
     }
 }
@@ -87,7 +87,7 @@ impl EngineApiMetrics {
 /// Metrics for the entire blockchain tree
 #[derive(Metrics)]
 #[metrics(scope = "blockchain_tree")]
-pub struct TreeMetrics {
+pub(crate) struct TreeMetrics {
     /// The highest block number in the canonical chain
     pub canonical_chain_height: Gauge,
     /// The number of reorgs
@@ -103,7 +103,7 @@ pub struct TreeMetrics {
 /// Metrics for the `EngineApi`.
 #[derive(Metrics)]
 #[metrics(scope = "consensus.engine.beacon")]
-pub struct EngineMetrics {
+pub(crate) struct EngineMetrics {
     /// Engine API forkchoiceUpdated response type metrics
     #[metric(skip)]
     pub(crate) forkchoice_updated: ForkchoiceUpdatedMetrics,
@@ -336,42 +336,42 @@ pub(crate) struct BalMetrics {
 /// Metrics for non-execution related block validation.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.block_validation")]
-pub struct BlockValidationMetrics {
+pub(crate) struct BlockValidationMetrics {
     /// Total number of storage tries updated in the state root calculation
-    pub state_root_storage_tries_updated_total: Counter,
+    pub(crate) state_root_storage_tries_updated_total: Counter,
     /// Total number of times the parallel state root computation fell back to regular.
-    pub state_root_parallel_fallback_total: Counter,
+    pub(crate) state_root_parallel_fallback_total: Counter,
     /// Total number of times the state root task failed but the fallback succeeded.
-    pub state_root_task_fallback_success_total: Counter,
+    pub(crate) state_root_task_fallback_success_total: Counter,
     /// Latest state root duration, ie the time spent blocked waiting for the state root.
-    pub state_root_duration: Gauge,
+    pub(crate) state_root_duration: Gauge,
     /// Histogram for state root duration ie the time spent blocked waiting for the state root
-    pub state_root_histogram: Histogram,
+    pub(crate) state_root_histogram: Histogram,
     /// Histogram of deferred trie computation duration.
-    pub deferred_trie_compute_duration: Histogram,
+    pub(crate) deferred_trie_compute_duration: Histogram,
     /// Payload conversion and validation latency
-    pub payload_validation_duration: Gauge,
+    pub(crate) payload_validation_duration: Gauge,
     /// Histogram of payload validation latency
-    pub payload_validation_histogram: Histogram,
+    pub(crate) payload_validation_histogram: Histogram,
     /// Payload processor spawning duration
-    pub spawn_payload_processor: Histogram,
+    pub(crate) spawn_payload_processor: Histogram,
     /// Post-execution validation duration
-    pub post_execution_validation_duration: Histogram,
+    pub(crate) post_execution_validation_duration: Histogram,
     /// Total duration of the new payload call
-    pub total_duration: Histogram,
+    pub(crate) total_duration: Histogram,
     /// Size of `HashedPostStateSorted` (`total_len`)
-    pub hashed_post_state_size: Histogram,
+    pub(crate) hashed_post_state_size: Histogram,
     /// Size of `TrieUpdatesSorted` (`total_len`)
-    pub trie_updates_sorted_size: Histogram,
+    pub(crate) trie_updates_sorted_size: Histogram,
     /// Size of `AnchoredTrieInput` overlay `TrieUpdatesSorted` (`total_len`)
-    pub anchored_overlay_trie_updates_size: Histogram,
+    pub(crate) anchored_overlay_trie_updates_size: Histogram,
     /// Size of `AnchoredTrieInput` overlay `HashedPostStateSorted` (`total_len`)
-    pub anchored_overlay_hashed_state_size: Histogram,
+    pub(crate) anchored_overlay_hashed_state_size: Histogram,
 }
 
 impl BlockValidationMetrics {
     /// Records a new state root time, updating both the histogram and state root gauge
-    pub fn record_state_root(&self, trie_output: &TrieUpdates, elapsed_as_secs: f64) {
+    pub(crate) fn record_state_root(&self, trie_output: &TrieUpdates, elapsed_as_secs: f64) {
         self.state_root_storage_tries_updated_total
             .increment(trie_output.storage_tries_ref().len() as u64);
         self.state_root_duration.set(elapsed_as_secs);
@@ -380,7 +380,7 @@ impl BlockValidationMetrics {
 
     /// Records a new payload validation time, updating both the histogram and the payload
     /// validation gauge
-    pub fn record_payload_validation(&self, elapsed_as_secs: f64) {
+    pub(crate) fn record_payload_validation(&self, elapsed_as_secs: f64) {
         self.payload_validation_duration.set(elapsed_as_secs);
         self.payload_validation_histogram.record(elapsed_as_secs);
     }

crates/engine/tree/src/tree/mod.rs

@@ -3,7 +3,7 @@ use crate::{
     chain::FromOrchestrator,
     engine::{DownloadRequest, EngineApiEvent, EngineApiKind, EngineApiRequest, FromEngine},
     persistence::PersistenceHandle,
-    tree::{error::InsertPayloadError, payload_validator::TreeCtx},
+    tree::{error::InsertPayloadError, metrics::EngineApiMetrics, payload_validator::TreeCtx},
 };
 use alloy_consensus::BlockHeader;
 use alloy_eips::{eip1898::BlockWithParent, merge::EPOCH_SLOTS, BlockNumHash, NumHash};
@@ -30,7 +30,7 @@ use reth_payload_primitives::{
 };
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
 use reth_provider::{
-    BlockExecutionOutput, BlockExecutionResult, BlockReader, ChangeSetReader,
+    BlockExecutionOutput, BlockExecutionResult, BlockNumReader, BlockReader, ChangeSetReader,
     DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StageCheckpointReader,
     StateProviderBox, StateProviderFactory, StateReader, StorageChangeSetReader,
     TransactionVariant,
@@ -55,7 +55,7 @@ pub mod error;
 pub mod instrumented_state;
 mod invalid_headers;
 mod metrics;
-pub mod payload_processor;
+mod payload_processor;
 pub mod payload_validator;
 mod persistence_state;
 pub mod precompile_cache;
@@ -66,9 +66,7 @@ mod trie_updates;
 
 use crate::tree::error::AdvancePersistenceError;
 pub use block_buffer::BlockBuffer;
-pub use cached_state::{CachedStateMetrics, CachedStateProvider, ExecutionCache, SavedCache};
 pub use invalid_headers::InvalidHeaderCache;
-pub use metrics::EngineApiMetrics;
 pub use payload_processor::*;
 pub use payload_validator::{BasicEngineValidator, EngineValidator};
 pub use persistence_state::PersistenceState;
@@ -160,16 +158,6 @@ impl<N: NodePrimitives> EngineApiTreeState<N> {
             forkchoice_state_tracker: ForkchoiceStateTracker::default(),
         }
     }
-
-    /// Returns a reference to the tree state.
-    pub const fn tree_state(&self) -> &TreeState<N> {
-        &self.tree_state
-    }
-
-    /// Returns true if the block has been marked as invalid.
-    pub fn has_invalid_header(&mut self, hash: &B256) -> bool {
-        self.invalid_headers.get(hash).is_some()
-    }
 }
 
 /// The outcome of a tree operation.
@@ -333,10 +321,11 @@ where
         + HashedPostStateProvider
         + Clone
         + 'static,
-    P::Provider: BlockReader<Block = N::Block, Header = N::BlockHeader>
+    <P as DatabaseProviderFactory>::Provider: BlockReader<Block = N::Block, Header = N::BlockHeader>
         + StageCheckpointReader
         + ChangeSetReader
-        + StorageChangeSetReader,
+        + StorageChangeSetReader
+        + BlockNumReader,
     C: ConfigureEvm<Primitives = N> + 'static,
     T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
     V: EngineValidator<T>,
@@ -2613,27 +2602,19 @@ where
         let block_num_hash = block_id.block;
         debug!(target: "engine::tree", block=?block_num_hash, parent = ?block_id.parent, "Inserting new block into tree");
 
-        // Check if block already exists - first in memory, then DB only if it could be persisted
-        if self.state.tree_state.sealed_header_by_hash(&block_num_hash.hash).is_some() {
-            convert_to_block(self, input)?;
-            return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid));
-        }
-
-        // Only query DB if block could be persisted (number <= last persisted block).
-        // New blocks from CL always have number > last persisted, so skip DB lookup for them.
-        if block_num_hash.number <= self.persistence_state.last_persisted_block.number {
-            match self.provider.sealed_header_by_hash(block_num_hash.hash) {
-                Err(err) => {
-                    let block = convert_to_block(self, input)?;
-                    return Err(InsertBlockError::new(block, err.into()).into());
-                }
-                Ok(Some(_)) => {
-                    convert_to_block(self, input)?;
-                    return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid));
-                }
-                Ok(None) => {}
+        match self.sealed_header_by_hash(block_num_hash.hash) {
+            Err(err) => {
+                let block = convert_to_block(self, input)?;
+                return Err(InsertBlockError::new(block, err.into()).into());
             }
-        }
+            Ok(Some(_)) => {
+                // We now assume that we already have this block in the tree. However, we need to
+                // run the conversion to ensure that the block hash is valid.
+                convert_to_block(self, input)?;
+                return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid))
+            }
+            _ => {}
+        };
 
         // Ensure that the parent state is available.
         match self.state_provider_builder(block_id.parent) {

crates/engine/tree/src/tree/payload_processor/bal.rs

@@ -21,7 +21,7 @@ pub fn total_slots(bal: &BlockAccessList) -> usize {
 /// first, followed by read-only slots. The iterator intelligently skips accounts and slots
 /// outside the specified range for efficient traversal.
 #[derive(Debug)]
-pub struct BALSlotIter<'a> {
+pub(crate) struct BALSlotIter<'a> {
     bal: &'a BlockAccessList,
     range: Range<usize>,
     current_index: usize,
@@ -34,7 +34,7 @@ pub struct BALSlotIter<'a> {
 
 impl<'a> BALSlotIter<'a> {
     /// Creates a new iterator over storage slots within the specified range.
-    pub fn new(bal: &'a BlockAccessList, range: Range<usize>) -> Self {
+    pub(crate) fn new(bal: &'a BlockAccessList, range: Range<usize>) -> Self {
         let mut iter = Self { bal, range, current_index: 0, account_idx: 0, slot_idx: 0 };
         iter.skip_to_range_start();
         iter

crates/engine/tree/src/tree/payload_processor/mod.rs

@@ -7,7 +7,7 @@ use crate::tree::{
         prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
         sparse_trie::StateRootComputeOutcome,
     },
-    sparse_trie::{SparseTrieCacheTask, SparseTrieTask, SpawnedSparseTrieTask},
+    sparse_trie::{SparseTrieCacheTask, SparseTrieTask},
     StateProviderBuilder, TreeConfig,
 };
 use alloy_eip7928::BlockAccessList;
@@ -56,13 +56,10 @@ use tracing::{debug, debug_span, instrument, warn, Span};
 pub mod bal;
 pub mod executor;
 pub mod multiproof;
-mod preserved_sparse_trie;
 pub mod prewarm;
 pub mod receipt_root_task;
 pub mod sparse_trie;
 
-use preserved_sparse_trie::{PreservedSparseTrie, SharedPreservedSparseTrie};
-
 /// Default parallelism thresholds to use with the [`ParallelSparseTrie`].
 ///
 /// These values were determined by performing benchmarks using gradually increasing values to judge
@@ -125,16 +122,13 @@ where
     precompile_cache_disabled: bool,
     /// Precompile cache map.
     precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
-    /// A pruned `SparseStateTrie`, kept around as a cache of already revealed trie nodes and to
-    /// re-use allocated memory. Stored with the block hash it was computed for to enable trie
-    /// preservation across sequential payload validations.
-    sparse_state_trie: SharedPreservedSparseTrie,
+    /// A cleared `SparseStateTrie`, kept around to be reused for the state root computation so
+    /// that allocations can be minimized.
+    sparse_state_trie: Arc<
+        parking_lot::Mutex<Option<ClearedSparseStateTrie<ParallelSparseTrie, ParallelSparseTrie>>>,
+    >,
     /// Maximum concurrency for prewarm task.
     prewarm_max_concurrency: usize,
-    /// Sparse trie prune depth.
-    sparse_trie_prune_depth: usize,
-    /// Maximum storage tries to retain after pruning.
-    sparse_trie_max_storage_tries: usize,
     /// Whether to disable cache metrics recording.
     disable_cache_metrics: bool,
 }
@@ -145,7 +139,7 @@ where
     Evm: ConfigureEvm<Primitives = N>,
 {
     /// Returns a reference to the workload executor driving payload tasks.
-    pub const fn executor(&self) -> &WorkloadExecutor {
+    pub(super) const fn executor(&self) -> &WorkloadExecutor {
         &self.executor
     }
 
@@ -166,10 +160,8 @@ where
             disable_state_cache: config.disable_state_cache(),
             precompile_cache_disabled: config.precompile_cache_disabled(),
             precompile_cache_map,
-            sparse_state_trie: SharedPreservedSparseTrie::default(),
+            sparse_state_trie: Arc::default(),
             prewarm_max_concurrency: config.prewarm_max_concurrency(),
-            sparse_trie_prune_depth: config.sparse_trie_prune_depth(),
-            sparse_trie_max_storage_tries: config.sparse_trie_max_storage_tries(),
             disable_cache_metrics: config.disable_cache_metrics(),
         }
     }
@@ -235,7 +227,8 @@ where
             + 'static,
     {
         // start preparing transactions immediately
-        let (prewarm_rx, execution_rx) = self.spawn_tx_iterator(transactions);
+        let (prewarm_rx, execution_rx, transaction_count_hint) =
+            self.spawn_tx_iterator(transactions);
 
         let span = Span::current();
         let (to_sparse_trie, sparse_trie_rx) = channel();
@@ -244,9 +237,6 @@ where
         // Extract V2 proofs flag early so we can pass it to prewarm
         let v2_proofs_enabled = !config.disable_proof_v2();
 
-        // Capture parent_state_root before env is moved into spawn_caching_with
-        let parent_state_root = env.parent_state_root;
-
         // Handle BAL-based optimization if available
         let prewarm_handle = if let Some(bal) = bal {
             // When BAL is present, use BAL prewarming and send BAL to multiproof
@@ -259,6 +249,7 @@ where
             self.spawn_caching_with(
                 env,
                 prewarm_rx,
+                transaction_count_hint,
                 provider_builder.clone(),
                 None, // Don't send proof targets when BAL is present
                 Some(bal),
@@ -269,6 +260,7 @@ where
             self.spawn_caching_with(
                 env,
                 prewarm_rx,
+                transaction_count_hint,
                 provider_builder.clone(),
                 Some(to_multi_proof.clone()),
                 None,
@@ -326,7 +318,6 @@ where
             state_root_tx,
             from_multi_proof,
             config,
-            parent_state_root,
         );
 
         PayloadHandle {
@@ -342,7 +333,7 @@ where
     ///
     /// Returns a [`PayloadHandle`] to communicate with the task.
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
-    pub fn spawn_cache_exclusive<P, I: ExecutableTxIterator<Evm>>(
+    pub(super) fn spawn_cache_exclusive<P, I: ExecutableTxIterator<Evm>>(
         &self,
         env: ExecutionEnv<Evm>,
         transactions: I,
@@ -352,10 +343,10 @@ where
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
-        let (prewarm_rx, execution_rx) = self.spawn_tx_iterator(transactions);
+        let (prewarm_rx, execution_rx, size_hint) = self.spawn_tx_iterator(transactions);
         // This path doesn't use multiproof, so V2 proofs flag doesn't matter
         let prewarm_handle =
-            self.spawn_caching_with(env, prewarm_rx, provider_builder, None, bal, false);
+            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None, bal, false);
         PayloadHandle {
             to_multi_proof: None,
             prewarm_handle,
@@ -373,15 +364,19 @@ where
     ) -> (
         mpsc::Receiver<WithTxEnv<TxEnvFor<Evm>, I::Recovered>>,
         mpsc::Receiver<Result<WithTxEnv<TxEnvFor<Evm>, I::Recovered>, I::Error>>,
+        usize,
     ) {
+        let (transactions, convert) = transactions.into();
+        let transactions = transactions.into_par_iter();
+        let transaction_count_hint = transactions.len();
+
         let (ooo_tx, ooo_rx) = mpsc::channel();
         let (prewarm_tx, prewarm_rx) = mpsc::channel();
         let (execute_tx, execute_rx) = mpsc::channel();
 
         // Spawn a task that `convert`s all transactions in parallel and sends them out-of-order.
-        rayon::spawn(move || {
-            let (transactions, convert) = transactions.into();
-            transactions.into_par_iter().enumerate().for_each_with(ooo_tx, |ooo_tx, (idx, tx)| {
+        self.executor.spawn_blocking(move || {
+            transactions.enumerate().for_each_with(ooo_tx, |ooo_tx, (idx, tx)| {
                 let tx = convert(tx);
                 let tx = tx.map(|tx| {
                     let (tx_env, tx) = tx.into_parts();
@@ -417,14 +412,16 @@ where
             }
         });
 
-        (prewarm_rx, execute_rx)
+        (prewarm_rx, execute_rx, transaction_count_hint)
     }
 
     /// Spawn prewarming optionally wired to the multiproof task for target updates.
+    #[expect(clippy::too_many_arguments)]
     fn spawn_caching_with<P>(
         &self,
         env: ExecutionEnv<Evm>,
         mut transactions: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
+        transaction_count_hint: usize,
         provider_builder: StateProviderBuilder<N, P>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
         bal: Option<Arc<BlockAccessList>>,
@@ -459,6 +456,7 @@ where
             self.execution_cache.clone(),
             prewarm_ctx,
             to_multi_proof,
+            transaction_count_hint,
             self.prewarm_max_concurrency,
         );
 
@@ -499,8 +497,6 @@ where
     }
 
     /// Spawns the [`SparseTrieTask`] for this payload processor.
-    ///
-    /// The trie is preserved when the new payload is a child of the previous one.
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
     fn spawn_sparse_trie_task(
         &self,
@@ -509,111 +505,68 @@ where
         state_root_tx: mpsc::Sender<Result<StateRootComputeOutcome, ParallelStateRootError>>,
         from_multi_proof: CrossbeamReceiver<MultiProofMessage>,
         config: &TreeConfig,
-        parent_state_root: B256,
     ) {
-        let preserved_sparse_trie = self.sparse_state_trie.clone();
+        let cleared_sparse_trie = Arc::clone(&self.sparse_state_trie);
         let trie_metrics = self.trie_metrics.clone();
         let span = Span::current();
         let disable_sparse_trie_as_cache = !config.enable_sparse_trie_as_cache();
-        let prune_depth = self.sparse_trie_prune_depth;
-        let max_storage_tries = self.sparse_trie_max_storage_tries;
 
         self.executor.spawn_blocking(move || {
             let _enter = span.entered();
 
-            // Reuse a stored SparseStateTrie if available, applying continuation logic.
-            // If this payload's parent state root matches the preserved trie's anchor,
-            // we can reuse the pruned trie structure. Otherwise, we clear the trie but
-            // keep allocations.
-            let sparse_state_trie = preserved_sparse_trie
-                .take()
-                .map(|preserved| preserved.into_trie_for(parent_state_root))
-                .unwrap_or_else(|| {
-                    debug!(
-                        target: "engine::tree::payload_processor",
-                        "Creating new sparse trie - no preserved trie available"
-                    );
-                    let default_trie = RevealableSparseTrie::blind_from(
-                        ParallelSparseTrie::default().with_parallelism_thresholds(
-                            PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS,
-                        ),
-                    );
+            // Reuse a stored SparseStateTrie, or create a new one using the desired configuration
+            // if there's none to reuse.
+            let sparse_state_trie = cleared_sparse_trie.lock().take().unwrap_or_else(|| {
+                let default_trie = RevealableSparseTrie::blind_from(
+                    ParallelSparseTrie::default()
+                        .with_parallelism_thresholds(PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS),
+                );
+                ClearedSparseStateTrie::from_state_trie(
                     SparseStateTrie::new()
                         .with_accounts_trie(default_trie.clone())
                         .with_default_storage_trie(default_trie)
-                        .with_updates(true)
-                });
+                        .with_updates(true),
+                )
+            });
 
-            let mut task = if disable_sparse_trie_as_cache {
-                SpawnedSparseTrieTask::Cleared(SparseTrieTask::new(
+            let (result, trie) = if disable_sparse_trie_as_cache {
+                SparseTrieTask::new_with_cleared_trie(
                     sparse_trie_rx,
                     proof_worker_handle,
-                    trie_metrics.clone(),
+                    trie_metrics,
                     sparse_state_trie,
-                ))
+                )
+                .run()
             } else {
-                SpawnedSparseTrieTask::Cached(SparseTrieCacheTask::new_with_cleared_trie(
+                SparseTrieCacheTask::new_with_cleared_trie(
                     from_multi_proof,
                     proof_worker_handle,
-                    trie_metrics.clone(),
-                    ClearedSparseStateTrie::from_state_trie(sparse_state_trie),
-                ))
+                    trie_metrics,
+                    sparse_state_trie,
+                )
+                .run()
             };
 
-            let result = task.run();
-            // Capture the computed state_root before sending the result
-            let computed_state_root = result.as_ref().ok().map(|outcome| outcome.state_root);
+            // Send state root computation result
+            let _ = state_root_tx.send(result);
 
-            // Acquire the guard before sending the result to prevent a race condition:
-            // Without this, the next block could start after send() but before store(),
-            // causing take() to return None and forcing it to create a new empty trie
-            // instead of reusing the preserved one. Holding the guard ensures the next
-            // block's take() blocks until we've stored the trie for reuse.
-            let mut guard = preserved_sparse_trie.lock();
-
-            // Send state root computation result - next block may start but will block on take()
-            if state_root_tx.send(result).is_err() {
-                // Receiver dropped - payload was likely invalid or cancelled.
-                // Clear the trie instead of preserving potentially invalid state.
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    "State root receiver dropped, clearing trie"
-                );
-                let trie = task.into_cleared_trie(
-                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
-                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
-                );
-                guard.store(PreservedSparseTrie::cleared(trie));
-                return;
-            }
-
-            // Only preserve the trie as anchored if computation succeeded.
-            // A failed computation may have left the trie in a partially updated state.
-            let _enter =
-                debug_span!(target: "engine::tree::payload_processor", "preserve").entered();
-            if let Some(state_root) = computed_state_root {
-                let start = std::time::Instant::now();
-                let trie = task.into_trie_for_reuse(
-                    prune_depth,
-                    max_storage_tries,
-                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
-                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
-                );
-                trie_metrics
-                    .into_trie_for_reuse_duration_histogram
-                    .record(start.elapsed().as_secs_f64());
-                guard.store(PreservedSparseTrie::anchored(trie, state_root));
+            // Clear the SparseStateTrie, shrink, and replace it back into the mutex _after_
+            // sending results to the next step, so that time spent clearing
+            // doesn't block the step after this one.
+            let _enter = debug_span!(target: "engine::tree::payload_processor", "clear").entered();
+            let mut cleared_trie = if disable_sparse_trie_as_cache {
+                ClearedSparseStateTrie::from_state_trie(trie)
             } else {
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    "State root computation failed, clearing trie"
-                );
-                let trie = task.into_cleared_trie(
-                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
-                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
-                );
-                guard.store(PreservedSparseTrie::cleared(trie));
-            }
+                ClearedSparseStateTrie::pruned(trie)
+            };
+
+            // Shrink the sparse trie so that we don't have ever increasing memory.
+            cleared_trie.shrink_to(
+                SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
+                SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
+            );
+
+            cleared_sparse_trie.lock().replace(cleared_trie);
         });
     }
 
@@ -624,7 +577,7 @@ where
     ///
     /// The cache enables subsequent blocks to reuse account, storage, and bytecode data without
     /// hitting the database, maintaining performance consistency.
-    pub fn on_inserted_executed_block(
+    pub(crate) fn on_inserted_executed_block(
         &self,
         block_with_parent: BlockWithParent,
         bundle_state: &BundleState,
@@ -721,19 +674,19 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
     }
 
     /// Returns a clone of the caches used by prewarming
-    pub fn caches(&self) -> Option<ExecutionCache> {
+    pub(super) fn caches(&self) -> Option<ExecutionCache> {
         self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.cache().clone())
     }
 
     /// Returns a clone of the cache metrics used by prewarming
-    pub fn cache_metrics(&self) -> Option<CachedStateMetrics> {
+    pub(super) fn cache_metrics(&self) -> Option<CachedStateMetrics> {
         self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.metrics().clone())
     }
 
     /// Terminates the pre-warming transaction processing.
     ///
     /// Note: This does not terminate the task yet.
-    pub fn stop_prewarming_execution(&self) {
+    pub(super) fn stop_prewarming_execution(&self) {
         self.prewarm_handle.stop_prewarming_execution()
     }
 
@@ -744,7 +697,7 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
     /// path without cloning the expensive `BundleState`.
     ///
     /// Returns a sender for the channel that should be notified on block validation success.
-    pub fn terminate_caching(
+    pub(super) fn terminate_caching(
         &mut self,
         execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
     ) -> Option<mpsc::Sender<()>> {
@@ -764,7 +717,7 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
 /// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the
 /// prewarm task without cloning the expensive `BundleState`.
 #[derive(Debug)]
-pub struct CacheTaskHandle<R> {
+pub(crate) struct CacheTaskHandle<R> {
     /// The shared cache the task operates with.
     saved_cache: Option<SavedCache>,
     /// Channel to the spawned prewarm task if any
@@ -775,7 +728,7 @@ impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
     /// Terminates the pre-warming transaction processing.
     ///
     /// Note: This does not terminate the task yet.
-    pub fn stop_prewarming_execution(&self) {
+    pub(super) fn stop_prewarming_execution(&self) {
         self.to_prewarm_task
             .as_ref()
             .map(|tx| tx.send(PrewarmTaskEvent::TerminateTransactionExecution).ok());
@@ -786,7 +739,7 @@ impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
     /// If the [`BlockExecutionOutput`] is provided it will update the shared cache using its
     /// bundle state. Using `Arc<ExecutionOutcome>` avoids cloning the expensive `BundleState`.
     #[must_use = "sender must be used and notified on block validation success"]
-    pub fn terminate_caching(
+    pub(super) fn terminate_caching(
         &mut self,
         execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
     ) -> Option<mpsc::Sender<()>> {
@@ -840,7 +793,7 @@ impl<R> Drop for CacheTaskHandle<R> {
 /// - Prepares data for state root proof computation
 /// - Runs concurrently but must not interfere with cache saves
 #[derive(Clone, Debug, Default)]
-pub struct PayloadExecutionCache {
+struct PayloadExecutionCache {
     /// Guarded cloneable cache identified by a block hash.
     inner: Arc<RwLock<Option<SavedCache>>>,
     /// Metrics for cache operations.
@@ -920,7 +873,7 @@ impl PayloadExecutionCache {
     ///
     /// Violating this requirement can result in cache corruption, incorrect state data,
     /// and potential consensus failures.
-    pub fn update_with_guard<F>(&self, update_fn: F)
+    pub(crate) fn update_with_guard<F>(&self, update_fn: F)
     where
         F: FnOnce(&mut Option<SavedCache>),
     {
@@ -947,14 +900,6 @@ pub struct ExecutionEnv<Evm: ConfigureEvm> {
     pub hash: B256,
     /// Hash of the parent block.
     pub parent_hash: B256,
-    /// State root of the parent block.
-    /// Used for sparse trie continuation: if the preserved trie's anchor matches this,
-    /// the trie can be reused directly.
-    pub parent_state_root: B256,
-    /// Number of transactions in the block.
-    /// Used to determine parallel worker count for prewarming.
-    /// A value of 0 indicates the count is unknown.
-    pub transaction_count: usize,
 }
 
 impl<Evm: ConfigureEvm> Default for ExecutionEnv<Evm>
@@ -966,8 +911,6 @@ where
             evm_env: Default::default(),
             hash: Default::default(),
             parent_hash: Default::default(),
-            parent_state_root: Default::default(),
-            transaction_count: 0,
         }
     }
 }

crates/engine/tree/src/tree/payload_processor/multiproof.rs

@@ -100,7 +100,7 @@ impl SparseTrieUpdate {
 
 /// Messages used internally by the multi proof task.
 #[derive(Debug)]
-pub enum MultiProofMessage {
+pub(super) enum MultiProofMessage {
     /// Prefetch proof targets
     PrefetchProofs(VersionedMultiProofTargets),
     /// New state update from transaction execution with its source
@@ -257,7 +257,7 @@ fn extend_multiproof_targets(dest: &mut MultiProofTargets, src: &VersionedMultiP
 
 /// A set of multiproof targets which can be either in the legacy or V2 representations.
 #[derive(Debug)]
-pub enum VersionedMultiProofTargets {
+pub(super) enum VersionedMultiProofTargets {
     /// Legacy targets
     Legacy(MultiProofTargets),
     /// V2 targets
@@ -587,8 +587,6 @@ pub(crate) struct MultiProofTaskMetrics {
     pub first_update_wait_time_histogram: Histogram,
     /// Total time spent waiting for the last proof result.
     pub last_proof_wait_time_histogram: Histogram,
-    /// Time spent preparing the sparse trie for reuse after state root computation.
-    pub into_trie_for_reuse_duration_histogram: Histogram,
 }
 
 /// Standalone task that receives a transaction state stream and updates relevant

crates/engine/tree/src/tree/payload_processor/preserved_sparse_trie.rs

@@ -1,117 +0,0 @@
-//! Preserved sparse trie for reuse across payload validations.
-
-use alloy_primitives::B256;
-use parking_lot::Mutex;
-use reth_trie_sparse::SparseStateTrie;
-use reth_trie_sparse_parallel::ParallelSparseTrie;
-use std::sync::Arc;
-use tracing::debug;
-
-/// Type alias for the sparse trie type used in preservation.
-pub(super) type SparseTrie = SparseStateTrie<ParallelSparseTrie, ParallelSparseTrie>;
-
-/// Shared handle to a preserved sparse trie that can be reused across payload validations.
-///
-/// This is stored in [`PayloadProcessor`](super::PayloadProcessor) and cloned to pass to
-/// [`SparseTrieTask`](super::sparse_trie::SparseTrieTask) for trie reuse.
-#[derive(Debug, Default, Clone)]
-pub(super) struct SharedPreservedSparseTrie(Arc<Mutex<Option<PreservedSparseTrie>>>);
-
-impl SharedPreservedSparseTrie {
-    /// Takes the preserved trie if present, leaving `None` in its place.
-    pub(super) fn take(&self) -> Option<PreservedSparseTrie> {
-        self.0.lock().take()
-    }
-
-    /// Acquires a guard that blocks `take()` until dropped.
-    /// Use this before sending the state root result to ensure the next block
-    /// waits for the trie to be stored.
-    pub(super) fn lock(&self) -> PreservedTrieGuard<'_> {
-        PreservedTrieGuard(self.0.lock())
-    }
-}
-
-/// Guard that holds the lock on the preserved trie.
-/// While held, `take()` will block. Call `store()` to save the trie before dropping.
-pub(super) struct PreservedTrieGuard<'a>(parking_lot::MutexGuard<'a, Option<PreservedSparseTrie>>);
-
-impl PreservedTrieGuard<'_> {
-    /// Stores a preserved trie for later reuse.
-    pub(super) fn store(&mut self, trie: PreservedSparseTrie) {
-        self.0.replace(trie);
-    }
-}
-
-/// A preserved sparse trie that can be reused across payload validations.
-///
-/// The trie exists in one of two states:
-/// - **Anchored**: Has a computed state root and can be reused for payloads whose parent state root
-///   matches the anchor.
-/// - **Cleared**: Trie data has been cleared but allocations are preserved for reuse.
-#[derive(Debug)]
-pub(super) enum PreservedSparseTrie {
-    /// Trie with a computed state root that can be reused for continuation payloads.
-    Anchored {
-        /// The sparse state trie (pruned after root computation).
-        trie: SparseTrie,
-        /// The state root this trie represents (computed from the previous block).
-        /// Used to verify continuity: new payload's `parent_state_root` must match this.
-        state_root: B256,
-    },
-    /// Cleared trie with preserved allocations, ready for fresh use.
-    Cleared {
-        /// The sparse state trie with cleared data but preserved allocations.
-        trie: SparseTrie,
-    },
-}
-
-impl PreservedSparseTrie {
-    /// Creates a new anchored preserved trie.
-    ///
-    /// The `state_root` is the computed state root from the trie, which becomes the
-    /// anchor for determining if subsequent payloads can reuse this trie.
-    pub(super) const fn anchored(trie: SparseTrie, state_root: B256) -> Self {
-        Self::Anchored { trie, state_root }
-    }
-
-    /// Creates a cleared preserved trie (allocations preserved, data cleared).
-    pub(super) const fn cleared(trie: SparseTrie) -> Self {
-        Self::Cleared { trie }
-    }
-
-    /// Consumes self and returns the trie for reuse.
-    ///
-    /// If the preserved trie is anchored and the parent state root matches, the pruned
-    /// trie structure is reused directly. Otherwise, the trie is cleared but allocations
-    /// are preserved to reduce memory overhead.
-    pub(super) fn into_trie_for(self, parent_state_root: B256) -> SparseTrie {
-        match self {
-            Self::Anchored { trie, state_root } if state_root == parent_state_root => {
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    %state_root,
-                    "Reusing anchored sparse trie for continuation payload"
-                );
-                trie
-            }
-            Self::Anchored { mut trie, state_root } => {
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    anchor_root = %state_root,
-                    %parent_state_root,
-                    "Clearing anchored sparse trie - parent state root mismatch"
-                );
-                trie.clear();
-                trie
-            }
-            Self::Cleared { trie } => {
-                debug!(
-                    target: "engine::tree::payload_processor",
-                    %parent_state_root,
-                    "Using cleared sparse trie with preserved allocations"
-                );
-                trie
-            }
-        }
-    }
-}

crates/engine/tree/src/tree/payload_processor/prewarm.rs

@@ -49,8 +49,7 @@ use std::{
 use tracing::{debug, debug_span, instrument, trace, warn, Span};
 
 /// Determines the prewarming mode: transaction-based or BAL-based.
-#[derive(Debug)]
-pub enum PrewarmMode<Tx> {
+pub(super) enum PrewarmMode<Tx> {
     /// Prewarm by executing transactions from a stream.
     Transactions(Receiver<Tx>),
     /// Prewarm by prefetching slots from a Block Access List.
@@ -70,8 +69,7 @@ struct IndexedTransaction<Tx> {
 /// individually in parallel.
 ///
 /// Note: This task runs until cancelled externally.
-#[derive(Debug)]
-pub struct PrewarmCacheTask<N, P, Evm>
+pub(super) struct PrewarmCacheTask<N, P, Evm>
 where
     N: NodePrimitives,
     Evm: ConfigureEvm<Primitives = N>,
@@ -84,6 +82,8 @@ where
     ctx: PrewarmContext<N, P, Evm>,
     /// How many transactions should be executed in parallel
     max_concurrency: usize,
+    /// The number of transactions to be processed
+    transaction_count_hint: usize,
     /// Sender to emit evm state outcome messages, if any.
     to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
     /// Receiver for events produced by tx execution
@@ -99,11 +99,12 @@ where
     Evm: ConfigureEvm<Primitives = N> + 'static,
 {
     /// Initializes the task with the given transactions pending execution
-    pub fn new(
+    pub(super) fn new(
         executor: WorkloadExecutor,
         execution_cache: PayloadExecutionCache,
         ctx: PrewarmContext<N, P, Evm>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
+        transaction_count_hint: usize,
         max_concurrency: usize,
     ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
         let (actions_tx, actions_rx) = channel();
@@ -111,7 +112,7 @@ where
         trace!(
             target: "engine::tree::payload_processor::prewarm",
             max_concurrency,
-            transaction_count = ctx.env.transaction_count,
+            transaction_count_hint,
             "Initialized prewarm task"
         );
 
@@ -121,6 +122,7 @@ where
                 execution_cache,
                 ctx,
                 max_concurrency,
+                transaction_count_hint,
                 to_multi_proof,
                 actions_rx,
                 parent_span: Span::current(),
@@ -144,6 +146,7 @@ where
         let executor = self.executor.clone();
         let ctx = self.ctx.clone();
         let max_concurrency = self.max_concurrency;
+        let transaction_count_hint = self.transaction_count_hint;
         let span = Span::current();
 
         self.executor.spawn_blocking(move || {
@@ -151,14 +154,13 @@ where
 
             let (done_tx, done_rx) = mpsc::channel();
 
-            // When transaction_count is 0, it means the count is unknown. In this case, spawn
+            // When transaction_count_hint is 0, it means the count is unknown. In this case, spawn
             // max workers to handle potentially many transactions in parallel rather
             // than bottlenecking on a single worker.
-            let transaction_count = ctx.env.transaction_count;
-            let workers_needed = if transaction_count == 0 {
+            let workers_needed = if transaction_count_hint == 0 {
                 max_concurrency
             } else {
-                transaction_count.min(max_concurrency)
+                transaction_count_hint.min(max_concurrency)
             };
 
             // Spawn workers
@@ -368,8 +370,11 @@ where
         name = "prewarm and caching",
         skip_all
     )]
-    pub fn run<Tx>(self, mode: PrewarmMode<Tx>, actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>)
-    where
+    pub(super) fn run<Tx>(
+        self,
+        mode: PrewarmMode<Tx>,
+        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
+    ) where
         Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
     {
         // Spawn execution tasks based on mode
@@ -431,29 +436,23 @@ where
 
 /// Context required by tx execution tasks.
 #[derive(Debug, Clone)]
-pub struct PrewarmContext<N, P, Evm>
+pub(super) struct PrewarmContext<N, P, Evm>
 where
     N: NodePrimitives,
     Evm: ConfigureEvm<Primitives = N>,
 {
-    /// The execution environment.
-    pub env: ExecutionEnv<Evm>,
-    /// The EVM configuration.
-    pub evm_config: Evm,
-    /// The saved cache.
-    pub saved_cache: Option<SavedCache>,
+    pub(super) env: ExecutionEnv<Evm>,
+    pub(super) evm_config: Evm,
+    pub(super) saved_cache: Option<SavedCache>,
     /// Provider to obtain the state
-    pub provider: StateProviderBuilder<N, P>,
-    /// The metrics for the prewarm task.
-    pub metrics: PrewarmMetrics,
+    pub(super) provider: StateProviderBuilder<N, P>,
+    pub(super) metrics: PrewarmMetrics,
     /// An atomic bool that tells prewarm tasks to not start any more execution.
-    pub terminate_execution: Arc<AtomicBool>,
-    /// Whether the precompile cache is disabled.
-    pub precompile_cache_disabled: bool,
-    /// The precompile cache map.
-    pub precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
+    pub(super) terminate_execution: Arc<AtomicBool>,
+    pub(super) precompile_cache_disabled: bool,
+    pub(super) precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
     /// Whether V2 proof calculation is enabled.
-    pub v2_proofs_enabled: bool,
+    pub(super) v2_proofs_enabled: bool,
 }
 
 impl<N, P, Evm> PrewarmContext<N, P, Evm>
@@ -853,8 +852,7 @@ fn multiproof_targets_v2_from_state(state: EvmState) -> (VersionedMultiProofTarg
 ///
 /// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the main
 /// execution path without cloning the expensive `BundleState`.
-#[derive(Debug)]
-pub enum PrewarmTaskEvent<R> {
+pub(super) enum PrewarmTaskEvent<R> {
     /// Forcefully terminate all remaining transaction execution.
     TerminateTransactionExecution,
     /// Forcefully terminate the task on demand and update the shared cache with the given output
@@ -884,7 +882,7 @@ pub enum PrewarmTaskEvent<R> {
 /// Metrics for transactions prewarming.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.prewarm")]
-pub struct PrewarmMetrics {
+pub(crate) struct PrewarmMetrics {
     /// The number of transactions to prewarm
     pub(crate) transactions: Gauge,
     /// A histogram of the number of transactions to prewarm

crates/engine/tree/src/tree/payload_processor/sparse_trie.rs

@@ -5,15 +5,14 @@ use crate::tree::{
     payload_processor::multiproof::{MultiProofTaskMetrics, SparseTrieUpdate},
 };
 use alloy_primitives::B256;
-use alloy_rlp::{Decodable, Encodable};
+use alloy_rlp::Decodable;
 use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
-use rayon::iter::ParallelIterator;
+use rayon::iter::{ParallelBridge, ParallelIterator};
 use reth_errors::ProviderError;
-use reth_primitives_traits::{Account, ParallelBridgeBuffered};
+use reth_primitives_traits::Account;
 use reth_revm::state::EvmState;
 use reth_trie::{
     proof_v2::Target, updates::TrieUpdates, HashedPostState, Nibbles, TrieAccount, EMPTY_ROOT_HASH,
-    TRIE_ACCOUNT_RLP_MAX_SIZE,
 };
 use reth_trie_parallel::{
     proof_task::{
@@ -37,64 +36,6 @@ use std::{
 };
 use tracing::{debug, debug_span, instrument, trace};
 
-#[expect(clippy::large_enum_variant)]
-pub(super) enum SpawnedSparseTrieTask<BPF, A, S>
-where
-    BPF: TrieNodeProviderFactory + Send + Sync,
-    BPF::AccountNodeProvider: TrieNodeProvider + Send + Sync,
-    BPF::StorageNodeProvider: TrieNodeProvider + Send + Sync,
-    A: SparseTrie + SparseTrieExt + Send + Sync + Default,
-    S: SparseTrie + SparseTrieExt + Send + Sync + Default + Clone,
-{
-    Cleared(SparseTrieTask<BPF, A, S>),
-    Cached(SparseTrieCacheTask<A, S>),
-}
-
-impl<BPF, A, S> SpawnedSparseTrieTask<BPF, A, S>
-where
-    BPF: TrieNodeProviderFactory + Send + Sync + Clone,
-    BPF::AccountNodeProvider: TrieNodeProvider + Send + Sync,
-    BPF::StorageNodeProvider: TrieNodeProvider + Send + Sync,
-    A: SparseTrie + SparseTrieExt + Send + Sync + Default,
-    S: SparseTrie + SparseTrieExt + Send + Sync + Default + Clone,
-{
-    pub(super) fn run(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
-        match self {
-            Self::Cleared(task) => task.run(),
-            Self::Cached(task) => task.run(),
-        }
-    }
-
-    pub(super) fn into_trie_for_reuse(
-        self,
-        prune_depth: usize,
-        max_storage_tries: usize,
-        max_nodes_capacity: usize,
-        max_values_capacity: usize,
-    ) -> SparseStateTrie<A, S> {
-        match self {
-            Self::Cleared(task) => task.into_cleared_trie(max_nodes_capacity, max_values_capacity),
-            Self::Cached(task) => task.into_trie_for_reuse(
-                prune_depth,
-                max_storage_tries,
-                max_nodes_capacity,
-                max_values_capacity,
-            ),
-        }
-    }
-
-    pub(super) fn into_cleared_trie(
-        self,
-        max_nodes_capacity: usize,
-        max_values_capacity: usize,
-    ) -> SparseStateTrie<A, S> {
-        match self {
-            Self::Cleared(task) => task.into_cleared_trie(max_nodes_capacity, max_values_capacity),
-            Self::Cached(task) => task.into_cleared_trie(max_nodes_capacity, max_values_capacity),
-        }
-    }
-}
-
 /// A task responsible for populating the sparse trie.
 pub(super) struct SparseTrieTask<BPF, A = SerialSparseTrie, S = SerialSparseTrie>
 where
@@ -116,29 +57,46 @@ where
     BPF: TrieNodeProviderFactory + Send + Sync + Clone,
     BPF::AccountNodeProvider: TrieNodeProvider + Send + Sync,
     BPF::StorageNodeProvider: TrieNodeProvider + Send + Sync,
-    A: SparseTrie + SparseTrieExt + Send + Sync + Default,
-    S: SparseTrie + SparseTrieExt + Send + Sync + Default + Clone,
+    A: SparseTrie + Send + Sync + Default,
+    S: SparseTrie + Send + Sync + Default + Clone,
 {
-    /// Creates a new sparse trie task with the given trie.
-    pub(super) const fn new(
+    /// Creates a new sparse trie, pre-populating with a [`ClearedSparseStateTrie`].
+    pub(super) fn new_with_cleared_trie(
         updates: mpsc::Receiver<SparseTrieUpdate>,
         blinded_provider_factory: BPF,
         metrics: MultiProofTaskMetrics,
-        trie: SparseStateTrie<A, S>,
+        sparse_state_trie: ClearedSparseStateTrie<A, S>,
     ) -> Self {
-        Self { updates, metrics, trie, blinded_provider_factory }
+        Self { updates, metrics, trie: sparse_state_trie.into_inner(), blinded_provider_factory }
     }
 
-    /// Runs the sparse trie task to completion, computing the state root.
+    /// Runs the sparse trie task to completion.
     ///
-    /// Receives [`SparseTrieUpdate`]s until the channel is closed, applying each update
-    /// to the trie. Once all updates are processed, computes and returns the final state root.
+    /// This waits for new incoming [`SparseTrieUpdate`].
+    ///
+    /// This concludes once the last trie update has been received.
+    ///
+    /// # Returns
+    ///
+    /// - State root computation outcome.
+    /// - `SparseStateTrie` that needs to be cleared and reused to avoid reallocations.
     #[instrument(
         level = "debug",
         target = "engine::tree::payload_processor::sparse_trie",
         skip_all
     )]
-    pub(super) fn run(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
+    pub(super) fn run(
+        mut self,
+    ) -> (Result<StateRootComputeOutcome, ParallelStateRootError>, SparseStateTrie<A, S>) {
+        // run the main loop to completion
+        let result = self.run_inner();
+        (result, self.trie)
+    }
+
+    /// Inner function to run the sparse trie task to completion.
+    ///
+    /// See [`Self::run`] for more information.
+    fn run_inner(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
         let now = Instant::now();
 
         let mut num_iterations = 0;
@@ -188,29 +146,14 @@ where
 
         Ok(StateRootComputeOutcome { state_root, trie_updates })
     }
-
-    /// Clears and shrinks the trie, discarding all state.
-    ///
-    /// Use this when the payload was invalid or cancelled - we don't want to preserve
-    /// potentially invalid trie state, but we keep the allocations for reuse.
-    pub(super) fn into_cleared_trie(
-        mut self,
-        max_nodes_capacity: usize,
-        max_values_capacity: usize,
-    ) -> SparseStateTrie<A, S> {
-        self.trie.clear();
-        self.trie.shrink_to(max_nodes_capacity, max_values_capacity);
-        self.trie
-    }
 }
 
-/// Sparse trie task implementation that uses in-memory sparse trie data to schedule proof fetching.
 pub(super) struct SparseTrieCacheTask<A = SerialSparseTrie, S = SerialSparseTrie> {
     /// Sender for proof results.
     proof_result_tx: CrossbeamSender<ProofResultMessage>,
     /// Receiver for proof results directly from workers.
     proof_result_rx: CrossbeamReceiver<ProofResultMessage>,
-    /// Receives updates from execution and prewarming.
+    /// Receives updates from the state root task.
     updates: CrossbeamReceiver<MultiProofMessage>,
     /// `SparseStateTrie` used for computing the state root.
     trie: SparseStateTrie<A, S>,
@@ -225,8 +168,8 @@ pub(super) struct SparseTrieCacheTask<A = SerialSparseTrie, S = SerialSparseTrie
     /// Those are being moved into `account_updates` once storage roots
     /// are revealed and/or calculated.
     ///
-    /// Invariant: for each entry in `pending_account_updates` account must either be already
-    /// revealed in the trie or have an entry in `account_updates`.
+    /// Invariant: for each entry in `pending_account_updates` there is a corresponding
+    /// [`LeafUpdate::Touched`] entry in `account_updates`.
     ///
     /// Values can be either of:
     ///   - None: account had a storage update and is awaiting storage root calculation and/or
@@ -234,14 +177,6 @@ pub(super) struct SparseTrieCacheTask<A = SerialSparseTrie, S = SerialSparseTrie
     ///   - Some(_): account was changed/destroyed and is awaiting storage root calculation/reveal
     ///     to complete.
     pending_account_updates: B256Map<Option<Option<Account>>>,
-    /// Cache of account proof targets that were already fetched/requested from the proof workers.
-    /// account -> lowest `min_len` requested.
-    fetched_account_targets: B256Map<u8>,
-    /// Cache of storage proof targets that have already been fetched/requested from the proof
-    /// workers. account -> slot -> lowest `min_len` requested.
-    fetched_storage_targets: B256Map<B256Map<u8>>,
-    /// Reusable buffer for RLP encoding of accounts.
-    account_rlp_buf: Vec<u8>,
     /// Metrics for the sparse trie.
     metrics: MultiProofTaskMetrics,
 }
@@ -268,54 +203,37 @@ where
             account_updates: Default::default(),
             storage_updates: Default::default(),
             pending_account_updates: Default::default(),
-            fetched_account_targets: Default::default(),
-            fetched_storage_targets: Default::default(),
-            account_rlp_buf: Vec::with_capacity(TRIE_ACCOUNT_RLP_MAX_SIZE),
             metrics,
         }
     }
 
-    /// Prunes and shrinks the trie for reuse in the next payload built on top of this one.
-    ///
-    /// Should be called after the state root result has been sent.
-    pub(super) fn into_trie_for_reuse(
-        mut self,
-        prune_depth: usize,
-        max_storage_tries: usize,
-        max_nodes_capacity: usize,
-        max_values_capacity: usize,
-    ) -> SparseStateTrie<A, S> {
-        self.trie.prune(prune_depth, max_storage_tries);
-        self.trie.shrink_to(max_nodes_capacity, max_values_capacity);
-        self.trie
-    }
-
-    /// Clears and shrinks the trie, discarding all state.
-    ///
-    /// Use this when the payload was invalid or cancelled - we don't want to preserve
-    /// potentially invalid trie state, but we keep the allocations for reuse.
-    pub(super) fn into_cleared_trie(
-        mut self,
-        max_nodes_capacity: usize,
-        max_values_capacity: usize,
-    ) -> SparseStateTrie<A, S> {
-        self.trie.clear();
-        self.trie.shrink_to(max_nodes_capacity, max_values_capacity);
-        self.trie
-    }
-
     /// Runs the sparse trie task to completion.
     ///
-    /// This waits for new incoming [`MultiProofMessage`]s, applies updates to the trie and
-    /// schedules proof fetching when needed.
+    /// This waits for new incoming [`SparseTrieUpdate`].
     ///
-    /// This concludes once the last state update has been received and processed.
+    /// This concludes once the last trie update has been received.
+    ///
+    /// # Returns
+    ///
+    /// - State root computation outcome.
+    /// - `SparseStateTrie` that needs to be cleared and reused to avoid reallocations.
     #[instrument(
         level = "debug",
         target = "engine::tree::payload_processor::sparse_trie",
         skip_all
     )]
-    pub(super) fn run(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
+    pub(super) fn run(
+        mut self,
+    ) -> (Result<StateRootComputeOutcome, ParallelStateRootError>, SparseStateTrie<A, S>) {
+        // run the main loop to completion
+        let result = self.run_inner();
+        (result, self.trie)
+    }
+
+    /// Inner function to run the sparse trie task to completion.
+    ///
+    /// See [`Self::run`] for more information.
+    fn run_inner(&mut self) -> Result<StateRootComputeOutcome, ParallelStateRootError> {
         let now = Instant::now();
 
         let mut finished_state_updates = false;
@@ -335,6 +253,8 @@ where
                         }
                     };
 
+                    tracing::debug!("received update: {:?}", update);
+
                     match update {
                         MultiProofMessage::PrefetchProofs(targets) => {
                             self.on_prewarm_targets(targets);
@@ -353,8 +273,16 @@ where
                 }
             }
 
+            tracing::debug!("account updates: {:?}", self.account_updates);
+            tracing::debug!("storage updates: {:?}", self.storage_updates);
+            tracing::debug!("pending account updates: {:?}", self.pending_account_updates);
+
             self.process_updates()?;
 
+            tracing::debug!("account updates after: {:?}", self.account_updates);
+            tracing::debug!("storage updates after: {:?}", self.storage_updates);
+            tracing::debug!("pending account updates after: {:?}", self.pending_account_updates);
+
             if finished_state_updates &&
                 self.account_updates.is_empty() &&
                 self.storage_updates.iter().all(|(_, updates)| updates.is_empty())
@@ -463,7 +391,7 @@ where
         result: ProofResultMessage,
     ) -> Result<(), ParallelStateRootError> {
         let ProofResult::V2(result) = result.result? else {
-            unreachable!("sparse trie as cache must only be used with multiproof v2");
+            unreachable!("sparse trie as cache must only be used wit hmultiproof v2");
         };
 
         self.trie.reveal_decoded_multiproof_v2(result).map_err(|e| {
@@ -471,33 +399,19 @@ where
         })
     }
 
-    /// Applies updates to the sparse trie and dispatches requested multiproof targets.
+    /// Applies updates to the sparse trie and dispathes requested multiproof targets.
     fn process_updates(&mut self) -> Result<(), ProviderError> {
         let mut targets = MultiProofTargetsV2::default();
 
         for (addr, updates) in &mut self.storage_updates {
             let trie = self.trie.get_or_create_storage_trie_mut(*addr);
-            let fetched_storage = self.fetched_storage_targets.entry(*addr).or_default();
 
-            trie.update_leaves(updates, |path, min_len| match fetched_storage.entry(path) {
-                Entry::Occupied(mut entry) => {
-                    if min_len < *entry.get() {
-                        entry.insert(min_len);
-                        targets
-                            .storage_targets
-                            .entry(*addr)
-                            .or_default()
-                            .push(Target::new(path).with_min_len(min_len));
-                    }
-                }
-                Entry::Vacant(entry) => {
-                    entry.insert(min_len);
-                    targets
-                        .storage_targets
-                        .entry(*addr)
-                        .or_default()
-                        .push(Target::new(path).with_min_len(min_len));
-                }
+            trie.update_leaves(updates, |key, min_len| {
+                targets
+                    .storage_targets
+                    .entry(*addr)
+                    .or_default()
+                    .push(Target::new(key).with_min_len(min_len));
             })
             .map_err(ProviderError::other)?;
 
@@ -517,12 +431,10 @@ where
                     {
                         Vec::new()
                     } else {
-                        self.account_rlp_buf.clear();
-                        account
-                            .unwrap_or_default()
-                            .into_trie_account(storage_root)
-                            .encode(&mut self.account_rlp_buf);
-                        self.account_rlp_buf.clone()
+                        // TODO: optimize allocation
+                        alloy_rlp::encode(
+                            account.unwrap_or_default().into_trie_account(storage_root),
+                        )
                     };
                     self.account_updates.insert(*addr, LeafUpdate::Changed(encoded));
                 }
@@ -530,7 +442,6 @@ where
         }
 
         // Now handle pending account updates that can be upgraded to a proper update.
-        let account_rlp_buf = &mut self.account_rlp_buf;
         self.pending_account_updates.retain(|addr, account| {
             // If account has pending storage updates, it is still pending.
             if self.storage_updates.get(addr).is_some_and(|updates| !updates.is_empty()) {
@@ -550,7 +461,7 @@ where
             let trie_account = trie_account.map(|value| TrieAccount::decode(&mut &value[..]).expect("invalid account RLP"));
 
             let (account, storage_root) = if let Some(account) = account.take() {
-                // If account is Some(_) here it means it didn't have any storage updates
+                // If account is Some(_) here it means it didn't have any storage updates 
                 // and we can fetch the storage root directly from the account trie.
                 //
                 // If it did have storage updates, we would've had processed it above when iterating over storage tries.
@@ -558,40 +469,32 @@ where
 
                 (account, storage_root)
             } else {
-                (trie_account.map(Into::into), self.trie.storage_root(addr).expect("account had storage updates that were applied to its trie, storage root must be revealed by now"))
+                (trie_account.map(Into::into), self.trie.storage_root(addr).expect("account had storage updates that were applies to its trie, storage root must be revealed by now"))
             };
 
             let encoded = if account.is_none_or(|account| account.is_empty()) && storage_root == EMPTY_ROOT_HASH {
                 Vec::new()
             } else {
-                account_rlp_buf.clear();
-                account.unwrap_or_default().into_trie_account(storage_root).encode(account_rlp_buf);
-                account_rlp_buf.clone()
+                let account = account.unwrap_or_default().into_trie_account(storage_root);
+
+                // TODO: optimize allocation
+                alloy_rlp::encode(account)
             };
             self.account_updates.insert(*addr, LeafUpdate::Changed(encoded));
 
             false
         });
 
+        tracing::debug!("applying account updates: {:?}", self.account_updates);
         // Process account trie updates and fill the account targets.
         self.trie
             .trie_mut()
-            .update_leaves(&mut self.account_updates, |target, min_len| {
-                match self.fetched_account_targets.entry(target) {
-                    Entry::Occupied(mut entry) => {
-                        if min_len < *entry.get() {
-                            entry.insert(min_len);
-                            targets.account_targets.push(Target::new(target).with_min_len(min_len));
-                        }
-                    }
-                    Entry::Vacant(entry) => {
-                        entry.insert(min_len);
-                        targets.account_targets.push(Target::new(target).with_min_len(min_len));
-                    }
-                }
+            .update_leaves(&mut self.account_updates, |key, min_len| {
+                targets.account_targets.push(Target::new(key).with_min_len(min_len));
             })
             .map_err(ProviderError::other)?;
 
+        tracing::debug!("dispatching account multiproof with targets: {:?}", targets);
         if !targets.is_empty() {
             self.proof_worker_handle.dispatch_account_multiproof(AccountMultiproofInput::V2 {
                 targets,
@@ -657,7 +560,7 @@ where
         .storages
         .into_iter()
         .map(|(address, storage)| (address, storage, trie.take_storage_trie(&address)))
-        .par_bridge_buffered()
+        .par_bridge()
         .map(|(address, storage, storage_trie)| {
             let _enter =
                 debug_span!(target: "engine::tree::payload_processor::sparse_trie", parent: &span, "storage trie", ?address)

crates/engine/tree/src/tree/payload_validator.rs

@@ -402,13 +402,7 @@ where
             .in_scope(|| self.evm_env_for(&input))
             .map_err(NewPayloadError::other)?;
 
-        let env = ExecutionEnv {
-            evm_env,
-            hash: input.hash(),
-            parent_hash: input.parent_hash(),
-            parent_state_root: parent_block.state_root(),
-            transaction_count: input.transaction_count(),
-        };
+        let env = ExecutionEnv { evm_env, hash: input.hash(), parent_hash: input.parent_hash() };
 
         // Plan the strategy used for state root computation.
         let strategy = self.plan_state_root_computation();

crates/engine/tree/src/tree/precompile_cache.rs

@@ -21,8 +21,7 @@ impl<S> PrecompileCacheMap<S>
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
-    /// Get the precompile cache for the given address.
-    pub fn cache_for_address(&self, address: Address) -> PrecompileCache<S> {
+    pub(crate) fn cache_for_address(&self, address: Address) -> PrecompileCache<S> {
         // Try just using `.get` first to avoid acquiring a write lock.
         if let Some(cache) = self.0.get(&address) {
             return cache.clone();
@@ -91,7 +90,7 @@ impl<S> CacheEntry<S> {
 
 /// A cache for precompile inputs / outputs.
 #[derive(Debug)]
-pub struct CachedPrecompile<S>
+pub(crate) struct CachedPrecompile<S>
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
@@ -110,7 +109,7 @@ where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
     /// `CachedPrecompile` constructor.
-    pub const fn new(
+    pub(crate) const fn new(
         precompile: DynPrecompile,
         cache: PrecompileCache<S>,
         spec_id: S,
@@ -119,8 +118,7 @@ where
         Self { precompile, cache, spec_id, metrics }
     }
 
-    /// Wrap the given precompile in a cached precompile.
-    pub fn wrap(
+    pub(crate) fn wrap(
         precompile: DynPrecompile,
         cache: PrecompileCache<S>,
         spec_id: S,
@@ -198,18 +196,18 @@ where
 /// Metrics for the cached precompile.
 #[derive(reth_metrics::Metrics, Clone)]
 #[metrics(scope = "sync.caching")]
-pub struct CachedPrecompileMetrics {
+pub(crate) struct CachedPrecompileMetrics {
     /// Precompile cache hits
-    pub precompile_cache_hits: metrics::Counter,
+    precompile_cache_hits: metrics::Counter,
 
     /// Precompile cache misses
-    pub precompile_cache_misses: metrics::Counter,
+    precompile_cache_misses: metrics::Counter,
 
     /// Precompile cache size. Uses the LRU cache length as the size metric.
-    pub precompile_cache_size: metrics::Gauge,
+    precompile_cache_size: metrics::Gauge,
 
     /// Precompile execution errors.
-    pub precompile_errors: metrics::Counter,
+    precompile_errors: metrics::Counter,
 }
 
 impl CachedPrecompileMetrics {
@@ -217,7 +215,7 @@ impl CachedPrecompileMetrics {
     ///
     /// Adds address as an `address` label padded with zeros to at least two hex symbols, prefixed
     /// by `0x`.
-    pub fn new_with_address(address: Address) -> Self {
+    pub(crate) fn new_with_address(address: Address) -> Self {
         Self::new_with_labels(&[("address", format!("0x{address:02x}"))])
     }
 }

crates/engine/tree/src/tree/state.rs

@@ -48,7 +48,7 @@ pub struct TreeState<N: NodePrimitives = EthPrimitives> {
 
 impl<N: NodePrimitives> TreeState<N> {
     /// Returns a new, empty tree state that points to the given canonical head.
-    pub fn new(current_canonical_head: BlockNumHash, engine_kind: EngineApiKind) -> Self {
+    pub(crate) fn new(current_canonical_head: BlockNumHash, engine_kind: EngineApiKind) -> Self {
         Self {
             blocks_by_hash: HashMap::default(),
             blocks_by_number: BTreeMap::new(),
@@ -60,22 +60,25 @@ impl<N: NodePrimitives> TreeState<N> {
     }
 
     /// Resets the state and points to the given canonical head.
-    pub fn reset(&mut self, current_canonical_head: BlockNumHash) {
+    pub(crate) fn reset(&mut self, current_canonical_head: BlockNumHash) {
         *self = Self::new(current_canonical_head, self.engine_kind);
     }
 
     /// Returns the number of executed blocks stored.
-    pub fn block_count(&self) -> usize {
+    pub(crate) fn block_count(&self) -> usize {
         self.blocks_by_hash.len()
     }
 
     /// Returns the [`ExecutedBlock`] by hash.
-    pub fn executed_block_by_hash(&self, hash: B256) -> Option<&ExecutedBlock<N>> {
+    pub(crate) fn executed_block_by_hash(&self, hash: B256) -> Option<&ExecutedBlock<N>> {
         self.blocks_by_hash.get(&hash)
     }
 
     /// Returns the sealed block header by hash.
-    pub fn sealed_header_by_hash(&self, hash: &B256) -> Option<SealedHeader<N::BlockHeader>> {
+    pub(crate) fn sealed_header_by_hash(
+        &self,
+        hash: &B256,
+    ) -> Option<SealedHeader<N::BlockHeader>> {
         self.blocks_by_hash.get(hash).map(|b| b.sealed_block().sealed_header().clone())
     }
 
@@ -84,7 +87,7 @@ impl<N: NodePrimitives> TreeState<N> {
     /// highest persisted block connected to this chain.
     ///
     /// Returns `None` if the block for the given hash is not found.
-    pub fn blocks_by_hash(&self, hash: B256) -> Option<(B256, Vec<ExecutedBlock<N>>)> {
+    pub(crate) fn blocks_by_hash(&self, hash: B256) -> Option<(B256, Vec<ExecutedBlock<N>>)> {
         let block = self.blocks_by_hash.get(&hash).cloned()?;
         let mut parent_hash = block.recovered_block().parent_hash();
         let mut blocks = vec![block];
@@ -157,7 +160,7 @@ impl<N: NodePrimitives> TreeState<N> {
     }
 
     /// Insert executed block into the state.
-    pub fn insert_executed(&mut self, executed: ExecutedBlock<N>) {
+    pub(crate) fn insert_executed(&mut self, executed: ExecutedBlock<N>) {
         let hash = executed.recovered_block().hash();
         let parent_hash = executed.recovered_block().parent_hash();
         let block_number = executed.recovered_block().number();
@@ -213,7 +216,7 @@ impl<N: NodePrimitives> TreeState<N> {
     }
 
     /// Returns whether or not the hash is part of the canonical chain.
-    pub fn is_canonical(&self, hash: B256) -> bool {
+    pub(crate) fn is_canonical(&self, hash: B256) -> bool {
         let mut current_block = self.current_canonical_head.hash;
         if current_block == hash {
             return true
@@ -231,7 +234,11 @@ impl<N: NodePrimitives> TreeState<N> {
 
     /// Removes canonical blocks below the upper bound, only if the last persisted hash is
     /// part of the canonical chain.
-    pub fn remove_canonical_until(&mut self, upper_bound: BlockNumber, last_persisted_hash: B256) {
+    pub(crate) fn remove_canonical_until(
+        &mut self,
+        upper_bound: BlockNumber,
+        last_persisted_hash: B256,
+    ) {
         debug!(target: "engine::tree", ?upper_bound, ?last_persisted_hash, "Removing canonical blocks from the tree");
 
         // If the last persisted hash is not canonical, then we don't want to remove any canonical
@@ -256,7 +263,7 @@ impl<N: NodePrimitives> TreeState<N> {
 
     /// Removes all blocks that are below the finalized block, as well as removing non-canonical
     /// sidechains that fork from below the finalized block.
-    pub fn prune_finalized_sidechains(&mut self, finalized_num_hash: BlockNumHash) {
+    pub(crate) fn prune_finalized_sidechains(&mut self, finalized_num_hash: BlockNumHash) {
         let BlockNumHash { number: finalized_num, hash: finalized_hash } = finalized_num_hash;
 
         // We remove disconnected sidechains in three steps:
@@ -316,7 +323,7 @@ impl<N: NodePrimitives> TreeState<N> {
     /// NOTE: if the finalized block is greater than the upper bound, the only blocks that will be
     /// removed are canonical blocks and sidechains that fork below the `upper_bound`. This is the
     /// same behavior as if the `finalized_num` were `Some(upper_bound)`.
-    pub fn remove_until(
+    pub(crate) fn remove_until(
         &mut self,
         upper_bound: BlockNumHash,
         last_persisted_hash: B256,
@@ -354,22 +361,22 @@ impl<N: NodePrimitives> TreeState<N> {
     }
 
     /// Updates the canonical head to the given block.
-    pub const fn set_canonical_head(&mut self, new_head: BlockNumHash) {
+    pub(crate) const fn set_canonical_head(&mut self, new_head: BlockNumHash) {
         self.current_canonical_head = new_head;
     }
 
     /// Returns the tracked canonical head.
-    pub const fn canonical_head(&self) -> &BlockNumHash {
+    pub(crate) const fn canonical_head(&self) -> &BlockNumHash {
         &self.current_canonical_head
     }
 
     /// Returns the block hash of the canonical head.
-    pub const fn canonical_block_hash(&self) -> B256 {
+    pub(crate) const fn canonical_block_hash(&self) -> B256 {
         self.canonical_head().hash
     }
 
     /// Returns the block number of the canonical head.
-    pub const fn canonical_block_number(&self) -> BlockNumber {
+    pub(crate) const fn canonical_block_number(&self) -> BlockNumber {
         self.canonical_head().number
     }
 }
@@ -379,7 +386,7 @@ impl<N: NodePrimitives> TreeState<N> {
     /// Determines if the second block is a descendant of the first block.
     ///
     /// If the two blocks are the same, this returns `false`.
-    pub fn is_descendant(
+    pub(crate) fn is_descendant(
         &self,
         first: BlockNumHash,
         second: alloy_eips::eip1898::BlockWithParent,

crates/era-downloader/Cargo.toml

@@ -20,6 +20,8 @@ reth-era.workspace = true
 # http
 bytes.workspace = true
 reqwest.workspace = true
+reqwest.default-features = false
+reqwest.features = ["stream", "rustls-tls-native-roots"]
 
 # async
 tokio.workspace = true

crates/ethereum/cli/src/app.rs

@@ -86,7 +86,7 @@ where
         mut self,
         components: impl CliComponentsBuilder<N>,
         launcher: impl AsyncFnOnce(
-            WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+            WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
             Ext,
         ) -> Result<()>,
     ) -> Result<()>
@@ -132,7 +132,7 @@ pub(crate) fn run_commands_with<C, Ext, Rpc, N, SubCmd>(
     runner: CliRunner,
     components: impl CliComponentsBuilder<N>,
     launcher: impl AsyncFnOnce(
-        WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+        WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
         Ext,
     ) -> Result<()>,
 ) -> Result<()>
@@ -174,7 +174,7 @@ where
         }
         Commands::P2P(command) => runner.run_until_ctrl_c(command.execute::<N>()),
         Commands::Config(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::Prune(command) => runner.run_command_until_exit(|ctx| command.execute::<N>(ctx)),
+        Commands::Prune(command) => runner.run_until_ctrl_c(command.execute::<N>()),
         #[cfg(feature = "dev")]
         Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),
         Commands::ReExecute(command) => runner.run_until_ctrl_c(command.execute::<N>(components)),

crates/ethereum/cli/src/interface.rs

@@ -131,7 +131,7 @@ impl<
     /// ````
     pub fn run<L, Fut>(self, launcher: L) -> eyre::Result<()>
     where
-        L: FnOnce(WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>, Ext) -> Fut,
+        L: FnOnce(WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>, Ext) -> Fut,
         Fut: Future<Output = eyre::Result<()>>,
         C: ChainSpecParser<ChainSpec = ChainSpec>,
     {
@@ -148,7 +148,7 @@ impl<
         self,
         components: impl CliComponentsBuilder<N>,
         launcher: impl AsyncFnOnce(
-            WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+            WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
             Ext,
         ) -> eyre::Result<()>,
     ) -> eyre::Result<()>
@@ -180,7 +180,7 @@ impl<
     /// ```
     pub fn with_runner<L, Fut>(self, runner: CliRunner, launcher: L) -> eyre::Result<()>
     where
-        L: FnOnce(WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>, Ext) -> Fut,
+        L: FnOnce(WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>, Ext) -> Fut,
         Fut: Future<Output = eyre::Result<()>>,
         C: ChainSpecParser<ChainSpec = ChainSpec>,
     {
@@ -196,7 +196,7 @@ impl<
         runner: CliRunner,
         components: impl CliComponentsBuilder<N>,
         launcher: impl AsyncFnOnce(
-            WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>,
+            WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>,
             Ext,
         ) -> eyre::Result<()>,
     ) -> eyre::Result<()>

crates/ethereum/node/src/node.rs

@@ -119,9 +119,10 @@ impl EthereumNode {
     /// use reth_db::open_db_read_only;
     /// use reth_node_ethereum::EthereumNode;
     /// use reth_provider::providers::{RocksDBProvider, StaticFileProvider};
+    /// use std::sync::Arc;
     ///
     /// let factory = EthereumNode::provider_factory_builder()
-    ///     .db(open_db_read_only("db", Default::default()).unwrap())
+    ///     .db(Arc::new(open_db_read_only("db", Default::default()).unwrap()))
     ///     .chainspec(ChainSpecBuilder::mainnet().build().into())
     ///     .static_file(StaticFileProvider::read_only("db/static_files", false).unwrap())
     ///     .rocksdb_provider(RocksDBProvider::builder("db/rocksdb").build().unwrap())

crates/ethereum/node/tests/e2e/blobs.rs

@@ -100,12 +100,10 @@ async fn can_send_legacy_sidecar_post_activation() -> eyre::Result<()> {
         ChainSpecBuilder::default().chain(MAINNET.chain).genesis(genesis).osaka_activated().build(),
     );
     let genesis_hash = chain_spec.genesis_hash();
-    let node_config = NodeConfig::test().with_chain(chain_spec).with_unused_ports().with_rpc(
-        RpcServerArgs::default()
-            .with_unused_ports()
-            .with_http()
-            .with_force_blob_sidecar_upcasting(),
-    );
+    let node_config = NodeConfig::test()
+        .with_chain(chain_spec)
+        .with_unused_ports()
+        .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
     let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
         .testing_node(exec.clone())
         .node(EthereumNode::default())
@@ -127,7 +125,7 @@ async fn can_send_legacy_sidecar_post_activation() -> eyre::Result<()> {
     let blob_tx_hash = node.rpc.inject_tx(blob_tx).await?;
     // fetch it from rpc
     let envelope = node.rpc.envelope_by_hash(blob_tx_hash).await?;
-    // assert that sidecar was converted to eip7594 (force upcasting is enabled)
+    // assert that sidecar was converted to eip7594
     assert!(envelope.as_eip4844().unwrap().tx().sidecar().unwrap().is_eip7594());
     // validate sidecar
     TransactionTestContext::validate_sidecar(envelope);
@@ -163,12 +161,10 @@ async fn blob_conversion_at_osaka() -> eyre::Result<()> {
             .build(),
     );
     let genesis_hash = chain_spec.genesis_hash();
-    let node_config = NodeConfig::test().with_chain(chain_spec).with_unused_ports().with_rpc(
-        RpcServerArgs::default()
-            .with_unused_ports()
-            .with_http()
-            .with_force_blob_sidecar_upcasting(),
-    );
+    let node_config = NodeConfig::test()
+        .with_chain(chain_spec)
+        .with_unused_ports()
+        .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
     let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
         .testing_node(exec.clone())
         .node(EthereumNode::default())

crates/ethereum/node/tests/e2e/eth.rs

@@ -1,4 +1,4 @@
-use crate::utils::{advance_with_random_transactions, eth_payload_attributes};
+use crate::utils::eth_payload_attributes;
 use alloy_eips::eip7685::RequestsOrHash;
 use alloy_genesis::Genesis;
 use alloy_primitives::{Address, B256};
@@ -6,9 +6,8 @@ use alloy_rpc_types_engine::{PayloadAttributes, PayloadStatusEnum};
 use jsonrpsee_core::client::ClientT;
 use reth_chainspec::{ChainSpecBuilder, EthChainSpec, MAINNET};
 use reth_e2e_test_utils::{
-    node::NodeTestContext, setup, setup_engine, transaction::TransactionTestContext, wallet::Wallet,
+    node::NodeTestContext, setup, transaction::TransactionTestContext, wallet::Wallet,
 };
-use reth_node_api::TreeConfig;
 use reth_node_builder::{NodeBuilder, NodeHandle};
 use reth_node_core::{args::RpcServerArgs, node_config::NodeConfig};
 use reth_node_ethereum::EthereumNode;
@@ -257,56 +256,3 @@ async fn test_testing_build_block_v1_osaka() -> eyre::Result<()> {
 
     Ok(())
 }
-
-/// Tests that sparse trie allocation reuse works correctly across consecutive blocks.
-///
-/// This test exercises the sparse trie allocation reuse path by:
-/// 1. Starting a node with parallel state root computation enabled
-/// 2. Advancing multiple consecutive blocks with random transactions
-/// 3. Verifying that all blocks are successfully validated (state roots match)
-///
-/// Note: Trie structure reuse is currently disabled due to pruning creating blinded
-/// nodes. The preserved trie's allocations are still reused to reduce memory overhead,
-/// but the trie is cleared between blocks.
-#[tokio::test]
-async fn test_sparse_trie_reuse_across_blocks() -> eyre::Result<()> {
-    reth_tracing::init_test_tracing();
-
-    // Use parallel state root (non-legacy) with pruning enabled
-    let tree_config = TreeConfig::default()
-        .with_legacy_state_root(false)
-        .with_sparse_trie_prune_depth(2)
-        .with_sparse_trie_max_storage_tries(100);
-
-    let (mut nodes, _tasks, _wallet) = setup_engine::<EthereumNode>(
-        1,
-        Arc::new(
-            ChainSpecBuilder::default()
-                .chain(MAINNET.chain)
-                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
-                .cancun_activated()
-                .build(),
-        ),
-        false,
-        tree_config,
-        eth_payload_attributes,
-    )
-    .await?;
-
-    let mut node = nodes.pop().unwrap();
-
-    // Use a seeded RNG for reproducibility
-    let mut rng = rand::rng();
-
-    // Advance multiple consecutive blocks with random transactions.
-    // This exercises the sparse trie reuse path where each block's pruned trie
-    // is reused for the next block's state root computation.
-    let num_blocks = 5;
-    advance_with_random_transactions(&mut node, num_blocks, &mut rng, true).await?;
-
-    // Verify the chain advanced correctly
-    let best_block = node.inner.provider.best_block_number()?;
-    assert_eq!(best_block, num_blocks as u64, "Expected {} blocks, got {}", num_blocks, best_block);
-
-    Ok(())
-}

crates/ethereum/primitives/src/receipt.rs

@@ -511,8 +511,9 @@ mod compact {
             total_length += flags.len() + buffer.len();
             buf.put_slice(&flags);
             if zstd {
-                reth_zstd_compressors::with_receipt_compressor(|compressor| {
-                    let compressed = compressor.compress(&buffer).expect("Failed to compress.");
+                reth_zstd_compressors::RECEIPT_COMPRESSOR.with(|compressor| {
+                    let compressed =
+                        compressor.borrow_mut().compress(&buffer).expect("Failed to compress.");
                     buf.put(compressed.as_slice());
                 });
             } else {
@@ -524,7 +525,8 @@ mod compact {
         fn from_compact(buf: &[u8], _len: usize) -> (Self, &[u8]) {
             let (flags, mut buf) = ReceiptFlags::from(buf);
             if flags.__zstd() != 0 {
-                reth_zstd_compressors::with_receipt_decompressor(|decompressor| {
+                reth_zstd_compressors::RECEIPT_DECOMPRESSOR.with(|decompressor| {
+                    let decompressor = &mut decompressor.borrow_mut();
                     let decompressed = decompressor.decompress(buf);
                     let original_buf = buf;
                     let mut buf: &[u8] = decompressed;

crates/ethereum/primitives/src/transaction.rs

@@ -577,11 +577,19 @@ impl reth_codecs::Compact for TransactionSigned {
 
         let tx_bits = if zstd_bit {
             let mut tmp = Vec::with_capacity(256);
-            reth_zstd_compressors::with_tx_compressor(|compressor| {
+            if cfg!(feature = "std") {
+                reth_zstd_compressors::TRANSACTION_COMPRESSOR.with(|compressor| {
+                    let mut compressor = compressor.borrow_mut();
+                    let tx_bits = self.transaction.to_compact(&mut tmp);
+                    buf.put_slice(&compressor.compress(&tmp).expect("Failed to compress"));
+                    tx_bits as u8
+                })
+            } else {
+                let mut compressor = reth_zstd_compressors::create_tx_compressor();
                 let tx_bits = self.transaction.to_compact(&mut tmp);
                 buf.put_slice(&compressor.compress(&tmp).expect("Failed to compress"));
                 tx_bits as u8
-            })
+            }
         } else {
             self.transaction.to_compact(buf) as u8
         };
@@ -603,13 +611,26 @@ impl reth_codecs::Compact for TransactionSigned {
 
         let zstd_bit = bitflags >> 3;
         let (transaction, buf) = if zstd_bit != 0 {
-            reth_zstd_compressors::with_tx_decompressor(|decompressor| {
-                // TODO: enforce that zstd is only present at a "top" level type
+            if cfg!(feature = "std") {
+                reth_zstd_compressors::TRANSACTION_DECOMPRESSOR.with(|decompressor| {
+                    let mut decompressor = decompressor.borrow_mut();
+
+                    // TODO: enforce that zstd is only present at a "top" level type
+
+                    let transaction_type = (bitflags & 0b110) >> 1;
+                    let (transaction, _) =
+                        Transaction::from_compact(decompressor.decompress(buf), transaction_type);
+
+                    (transaction, buf)
+                })
+            } else {
+                let mut decompressor = reth_zstd_compressors::create_tx_decompressor();
                 let transaction_type = (bitflags & 0b110) >> 1;
                 let (transaction, _) =
                     Transaction::from_compact(decompressor.decompress(buf), transaction_type);
+
                 (transaction, buf)
-            })
+            }
         } else {
             let transaction_type = bitflags >> 1;
             Transaction::from_compact(buf, transaction_type)

crates/net/downloaders/src/file_client.rs

@@ -83,28 +83,6 @@ impl From<&'static str> for FileClientError {
 }
 
 impl<B: FullBlock> FileClient<B> {
-    /// Create a new file client from a slice of sealed blocks.
-    pub fn from_blocks(blocks: impl IntoIterator<Item = SealedBlock<B>>) -> Self {
-        let blocks: Vec<_> = blocks.into_iter().collect();
-        let capacity = blocks.len();
-
-        let mut headers = HashMap::with_capacity(capacity);
-        let mut hash_to_number = HashMap::with_capacity(capacity);
-        let mut bodies = HashMap::with_capacity(capacity);
-
-        for block in blocks {
-            let number = block.number();
-            let hash = block.hash();
-            let (header, body) = block.split_sealed_header_body();
-
-            headers.insert(number, header.into_header());
-            hash_to_number.insert(hash, number);
-            bodies.insert(hash, body);
-        }
-
-        Self { headers, hash_to_number, bodies }
-    }
-
     /// Create a new file client from a file path.
     pub async fn new<P: AsRef<Path>>(
         path: P,

crates/net/eth-wire/src/errors/p2p.rs

@@ -68,7 +68,7 @@ pub enum P2PStreamError {
     SendBufferFull,
 
     /// Disconnected error.
-    #[error("disconnected: {0}")]
+    #[error("disconnected")]
     Disconnected(DisconnectReason),
 
     /// Unknown disconnect reason error.

crates/node/builder/src/builder/mod.rs

@@ -251,8 +251,6 @@ impl<DB, ChainSpec: EthChainSpec> NodeBuilder<DB, ChainSpec> {
     }
 
     /// Creates a preconfigured node for testing purposes with a specific datadir.
-    ///
-    /// The entire `datadir` will be cleaned up when the node is dropped.
     #[cfg(feature = "test-utils")]
     pub fn testing_node_with_datadir(
         mut self,
@@ -270,7 +268,7 @@ impl<DB, ChainSpec: EthChainSpec> NodeBuilder<DB, ChainSpec> {
         let data_dir =
             path.unwrap_or_chain_default(self.config.chain.chain(), self.config.datadir.clone());
 
-        let db = reth_db::test_utils::create_test_rw_db_with_datadir(data_dir.data_dir());
+        let db = reth_db::test_utils::create_test_rw_db_with_path(data_dir.db());
 
         WithLaunchContext { builder: self.with_database(db), task_executor }
     }

crates/node/builder/src/launch/engine.rs

@@ -303,8 +303,20 @@ impl EngineNodeLauncher {
             // the CL
             loop {
                 tokio::select! {
-                    biased;
-
+                    shutdown_req = &mut shutdown_rx => {
+                        if let Ok(req) = shutdown_req {
+                            debug!(target: "reth::cli", "received engine shutdown request");
+                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(
+                                FromOrchestrator::Terminate { tx: req.done_tx }.into()
+                            );
+                        }
+                    }
+                    payload = built_payloads.select_next_some() => {
+                        if let Some(executed_block) = payload.executed_block() {
+                            debug!(target: "reth::cli", block=?executed_block.recovered_block.num_hash(),  "inserting built payload");
+                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(EngineApiRequest::InsertExecutedBlock(executed_block.into_executed_payload()).into());
+                        }
+                    }
                     event = engine_service.next() => {
                         let Some(event) = event else { break };
                         debug!(target: "reth::cli", "Event: {event}");
@@ -352,20 +364,6 @@ impl EngineNodeLauncher {
                             }
                         }
                     }
-                    payload = built_payloads.select_next_some() => {
-                        if let Some(executed_block) = payload.executed_block() {
-                            debug!(target: "reth::cli", block=?executed_block.recovered_block.num_hash(),  "inserting built payload");
-                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(EngineApiRequest::InsertExecutedBlock(executed_block.into_executed_payload()).into());
-                        }
-                    }
-                    shutdown_req = &mut shutdown_rx => {
-                        if let Ok(req) = shutdown_req {
-                            debug!(target: "reth::cli", "received engine shutdown request");
-                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(
-                                FromOrchestrator::Terminate { tx: req.done_tx }.into()
-                            );
-                        }
-                    }
                 }
             }
 

crates/node/builder/src/node.rs

@@ -218,9 +218,9 @@ impl<Node: FullNodeComponents, AddOns: NodeAddOns<Node>> DerefMut for FullNode<N
 }
 
 /// Helper type alias to define [`FullNode`] for a given [`Node`].
-pub type FullNodeFor<N, DB = DatabaseEnv> =
+pub type FullNodeFor<N, DB = Arc<DatabaseEnv>> =
     FullNode<NodeAdapter<RethFullAdapter<DB, N>>, <N as Node<RethFullAdapter<DB, N>>>::AddOns>;
 
 /// Helper type alias to define [`NodeHandle`] for a given [`Node`].
-pub type NodeHandleFor<N, DB = DatabaseEnv> =
+pub type NodeHandleFor<N, DB = Arc<DatabaseEnv>> =
     NodeHandle<NodeAdapter<RethFullAdapter<DB, N>>, <N as Node<RethFullAdapter<DB, N>>>::AddOns>;

crates/node/builder/src/rpc.rs

@@ -1192,7 +1192,6 @@ impl<'a, N: FullNodeComponents<Types: NodeTypes<ChainSpec: Hardforks + EthereumH
             .pending_block_kind(self.config.pending_block_kind)
             .raw_tx_forwarder(self.config.raw_tx_forwarder)
             .evm_memory_limit(self.config.rpc_evm_memory_limit)
-            .force_blob_sidecar_upcasting(self.config.force_blob_sidecar_upcasting)
     }
 }
 

crates/node/core/src/args/engine.rs

@@ -37,7 +37,6 @@ pub struct DefaultEngineValues {
     account_worker_count: Option<usize>,
     disable_proof_v2: bool,
     cache_metrics_disabled: bool,
-    enable_sparse_trie_as_cache: bool,
 }
 
 impl DefaultEngineValues {
@@ -173,12 +172,6 @@ impl DefaultEngineValues {
         self.cache_metrics_disabled = v;
         self
     }
-
-    /// Set whether to enable sparse trie as cache by default
-    pub const fn with_enable_sparse_trie_as_cache(mut self, v: bool) -> Self {
-        self.enable_sparse_trie_as_cache = v;
-        self
-    }
 }
 
 impl Default for DefaultEngineValues {
@@ -204,7 +197,6 @@ impl Default for DefaultEngineValues {
             account_worker_count: None,
             disable_proof_v2: false,
             cache_metrics_disabled: false,
-            enable_sparse_trie_as_cache: false,
         }
     }
 }
@@ -332,10 +324,6 @@ pub struct EngineArgs {
     /// Disable cache metrics recording, which can take up to 50ms with large cached state.
     #[arg(long = "engine.disable-cache-metrics", default_value_t = DefaultEngineValues::get_global().cache_metrics_disabled)]
     pub cache_metrics_disabled: bool,
-
-    /// Enable sparse trie as cache.
-    #[arg(long = "engine.enable-sparse-trie-as-cache", default_value_t = DefaultEngineValues::get_global().enable_sparse_trie_as_cache, conflicts_with = "disable_proof_v2")]
-    pub enable_sparse_trie_as_cache: bool,
 }
 
 #[allow(deprecated)]
@@ -362,7 +350,6 @@ impl Default for EngineArgs {
             account_worker_count,
             disable_proof_v2,
             cache_metrics_disabled,
-            enable_sparse_trie_as_cache,
         } = DefaultEngineValues::get_global().clone();
         Self {
             persistence_threshold,
@@ -389,7 +376,6 @@ impl Default for EngineArgs {
             account_worker_count,
             disable_proof_v2,
             cache_metrics_disabled,
-            enable_sparse_trie_as_cache,
         }
     }
 }
@@ -426,7 +412,6 @@ impl EngineArgs {
 
         config = config.with_disable_proof_v2(self.disable_proof_v2);
         config = config.without_cache_metrics(self.cache_metrics_disabled);
-        config = config.with_enable_sparse_trie_as_cache(self.enable_sparse_trie_as_cache);
 
         config
     }
@@ -479,7 +464,6 @@ mod tests {
             account_worker_count: Some(8),
             disable_proof_v2: false,
             cache_metrics_disabled: true,
-            enable_sparse_trie_as_cache: false,
         };
 
         let parsed_args = CommandParser::<EngineArgs>::parse_from([

crates/node/core/src/args/pruning.rs

@@ -6,7 +6,7 @@ use clap::{builder::RangedU64ValueParser, Args};
 use reth_chainspec::EthereumHardforks;
 use reth_config::config::PruneConfig;
 use reth_prune_types::{
-    PruneMode, PruneModes, ReceiptsLogPruneConfig, MINIMUM_DISTANCE, MINIMUM_UNWIND_SAFE_DISTANCE,
+    PruneMode, PruneModes, ReceiptsLogPruneConfig, MINIMUM_UNWIND_SAFE_DISTANCE,
 };
 use std::{collections::BTreeMap, ops::Not, sync::OnceLock};
 
@@ -81,7 +81,7 @@ impl Default for DefaultPruningValues {
             minimal_prune_modes: PruneModes {
                 sender_recovery: Some(PruneMode::Full),
                 transaction_lookup: Some(PruneMode::Full),
-                receipts: Some(PruneMode::Distance(MINIMUM_DISTANCE)),
+                receipts: Some(PruneMode::Full),
                 account_history: Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)),
                 storage_history: Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)),
                 bodies_history: Some(PruneMode::Distance(MINIMUM_UNWIND_SAFE_DISTANCE)),

crates/node/core/src/args/rpc_server.rs

@@ -645,16 +645,8 @@ pub struct RpcServerArgs {
     ///
     /// When enabled, transactions that fail execution will be skipped, and all subsequent
     /// transactions from the same sender will also be skipped.
-    #[arg(long = "testing.skip-invalid-transactions", default_value_t = true)]
+    #[arg(long = "testing.skip-invalid-transactions", default_value_t = false)]
     pub testing_skip_invalid_transactions: bool,
-
-    /// Force upcasting EIP-4844 blob sidecars to EIP-7594 format when Osaka is active.
-    ///
-    /// When enabled, blob transactions submitted via `eth_sendRawTransaction` with EIP-4844
-    /// sidecars will be automatically converted to EIP-7594 format if the next block is Osaka.
-    /// By default this is disabled, meaning transactions are submitted as-is.
-    #[arg(long = "rpc.force-blob-sidecar-upcasting", default_value_t = false)]
-    pub rpc_force_blob_sidecar_upcasting: bool,
 }
 
 impl RpcServerArgs {
@@ -776,12 +768,6 @@ impl RpcServerArgs {
         self.rpc_send_raw_transaction_sync_timeout = timeout;
         self
     }
-
-    /// Enables forced blob sidecar upcasting from EIP-4844 to EIP-7594 format.
-    pub const fn with_force_blob_sidecar_upcasting(mut self) -> Self {
-        self.rpc_force_blob_sidecar_upcasting = true;
-        self
-    }
 }
 
 impl Default for RpcServerArgs {
@@ -873,8 +859,7 @@ impl Default for RpcServerArgs {
             rpc_state_cache,
             gas_price_oracle,
             rpc_send_raw_transaction_sync_timeout,
-            testing_skip_invalid_transactions: true,
-            rpc_force_blob_sidecar_upcasting: false,
+            testing_skip_invalid_transactions: false,
         }
     }
 }
@@ -1051,7 +1036,6 @@ mod tests {
             },
             rpc_send_raw_transaction_sync_timeout: std::time::Duration::from_secs(30),
             testing_skip_invalid_transactions: true,
-            rpc_force_blob_sidecar_upcasting: false,
         };
 
         let parsed_args = CommandParser::<RpcServerArgs>::parse_from([

crates/node/ethstats/Cargo.toml

@@ -21,7 +21,7 @@ alloy-primitives.workspace = true
 alloy-consensus.workspace = true
 
 tokio.workspace = true
-tokio-tungstenite.workspace = true
+tokio-tungstenite = { workspace = true, features = ["rustls-tls-native-roots"] }
 futures-util.workspace = true
 tokio-stream.workspace = true
 

crates/node/ethstats/src/credentials.rs

@@ -13,23 +13,15 @@ pub(crate) struct EthstatsCredentials {
     pub secret: String,
     /// Host address of the `EthStats` server
     pub host: String,
-    /// Whether to use secure `WebSocket` (`WSS`) connection
-    pub use_tls: bool,
 }
 
 impl FromStr for EthstatsCredentials {
     type Err = EthStatsError;
 
-    /// Parse credentials from a string in the format "`node_id:secret@host`" or
-    /// "`node_id:secret@wss://host`"
-    ///
-    /// Supports the following formats:
-    /// - `node_id:secret@host` - Uses plain `WebSocket` (`ws://`)
-    /// - `node_id:secret@ws://host` - Explicitly use plain `WebSocket`
-    /// - `node_id:secret@wss://host` - Use secure `WebSocket` (`WSS`)
+    /// Parse credentials from a string in the format "`node_id:secret@host`"
     ///
     /// # Arguments
-    /// * `s` - String containing credentials
+    /// * `s` - String containing credentials in the format "`node_id:secret@host`"
     ///
     /// # Returns
     /// * `Ok(EthstatsCredentials)` - Successfully parsed credentials
@@ -40,7 +32,7 @@ impl FromStr for EthstatsCredentials {
             return Err(EthStatsError::InvalidUrl("Missing '@' separator".to_string()));
         }
         let creds = parts[0];
-        let mut host = parts[1].to_string();
+        let host = parts[1].to_string();
         let creds_parts: Vec<&str> = creds.split(':').collect();
         if creds_parts.len() != 2 {
             return Err(EthStatsError::InvalidUrl(
@@ -50,16 +42,6 @@ impl FromStr for EthstatsCredentials {
         let node_id = creds_parts[0].to_string();
         let secret = creds_parts[1].to_string();
 
-        // Detect and strip protocol prefix if present
-        let mut use_tls = false;
-        if let Some(stripped) = host.strip_prefix("wss://") {
-            use_tls = true;
-            host = stripped.to_string();
-        } else if let Some(stripped) = host.strip_prefix("ws://") {
-            use_tls = false;
-            host = stripped.to_string();
-        }
-
-        Ok(Self { node_id, secret, host, use_tls })
+        Ok(Self { node_id, secret, host })
     }
 }

crates/node/ethstats/src/ethstats.rs

@@ -102,15 +102,13 @@ where
     /// Establish `WebSocket` connection to the `EthStats` server
     ///
     /// Attempts to connect to the server using the credentials and handles
-    /// connection timeouts and errors. Uses either `ws://` or `wss://` based
-    /// on the credentials configuration.
+    /// connection timeouts and errors.
     async fn connect(&self) -> Result<(), EthStatsError> {
         debug!(
             target: "ethstats",
             "Attempting to connect to EthStats server at {}", self.credentials.host
         );
-        let protocol = if self.credentials.use_tls { "wss" } else { "ws" };
-        let full_url = format!("{}://{}/api", protocol, self.credentials.host);
+        let full_url = format!("ws://{}/api", self.credentials.host);
         let url = Url::parse(&full_url).map_err(EthStatsError::Url)?;
 
         match timeout(CONNECT_TIMEOUT, connect_async(url.as_str())).await {

crates/node/metrics/Cargo.toml

@@ -37,7 +37,7 @@ tempfile = { workspace = true, optional = true }
 tikv-jemalloc-ctl = { workspace = true, optional = true, features = ["stats"] }
 
 [target.'cfg(target_os = "linux")'.dependencies]
-procfs = "0.18.0"
+procfs = "0.17.0"
 
 [dev-dependencies]
 reqwest.workspace = true

crates/optimism/cli/src/app.rs

@@ -106,9 +106,7 @@ where
             }
             Commands::P2P(command) => runner.run_until_ctrl_c(command.execute::<OpNode>()),
             Commands::Config(command) => runner.run_until_ctrl_c(command.execute()),
-            Commands::Prune(command) => {
-                runner.run_command_until_exit(|ctx| command.execute::<OpNode>(ctx))
-            }
+            Commands::Prune(command) => runner.run_until_ctrl_c(command.execute::<OpNode>()),
             #[cfg(feature = "dev")]
             Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),
             Commands::ReExecute(command) => {

crates/optimism/cli/src/lib.rs

@@ -37,7 +37,7 @@ pub use commands::{import::ImportOpCommand, import_receipts::ImportReceiptsOpCom
 use reth_optimism_chainspec::OpChainSpec;
 use reth_rpc_server_types::{DefaultRpcModuleValidator, RpcModuleValidator};
 
-use std::{ffi::OsString, fmt, marker::PhantomData};
+use std::{ffi::OsString, fmt, marker::PhantomData, sync::Arc};
 
 use chainspec::OpChainSpecParser;
 use clap::Parser;
@@ -121,7 +121,7 @@ where
     /// [`NodeCommand`](reth_cli_commands::node::NodeCommand).
     pub fn run<L, Fut>(self, launcher: L) -> eyre::Result<()>
     where
-        L: FnOnce(WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>, Ext) -> Fut,
+        L: FnOnce(WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>, Ext) -> Fut,
         Fut: Future<Output = eyre::Result<()>>,
     {
         self.with_runner(CliRunner::try_default_runtime()?, launcher)
@@ -130,7 +130,7 @@ where
     /// Execute the configured cli command with the provided [`CliRunner`].
     pub fn with_runner<L, Fut>(self, runner: CliRunner, launcher: L) -> eyre::Result<()>
     where
-        L: FnOnce(WithLaunchContext<NodeBuilder<DatabaseEnv, C::ChainSpec>>, Ext) -> Fut,
+        L: FnOnce(WithLaunchContext<NodeBuilder<Arc<DatabaseEnv>, C::ChainSpec>>, Ext) -> Fut,
         Fut: Future<Output = eyre::Result<()>>,
     {
         let mut this = self.configure();

crates/optimism/flashblocks/Cargo.toml

@@ -38,7 +38,7 @@ op-alloy-rpc-types-engine = { workspace = true, features = ["k256"] }
 
 # io
 tokio.workspace = true
-tokio-tungstenite.workspace = true
+tokio-tungstenite = { workspace = true, features = ["rustls-tls-native-roots"] }
 serde_json.workspace = true
 url.workspace = true
 futures-util.workspace = true

crates/optimism/flashblocks/src/cache.rs

@@ -101,7 +101,7 @@ impl<T: SignedTransaction> SequenceManager<T> {
             // Bundle completed sequence with its decoded transactions and push to cache
             // Ring buffer automatically evicts oldest entry when full
             let txs = std::mem::take(&mut self.pending_transactions);
-            self.completed_cache.enqueue((completed, txs));
+            self.completed_cache.push((completed, txs));
 
             // ensure cache is wiped on new flashblock
             let _ = self.pending.take_cached_reads();

crates/optimism/flashblocks/tests/it/stream.rs

@@ -2,7 +2,6 @@ use futures_util::stream::StreamExt;
 use reth_optimism_flashblocks::WsFlashBlockStream;
 
 #[tokio::test]
-#[ignore = "depends on external service availability"]
 async fn test_streaming_flashblocks_from_remote_source_is_successful() {
     let items = 3;
     let ws_url = "wss://sepolia.flashblocks.base.org/ws".parse().unwrap();

crates/optimism/node/src/node.rs

@@ -219,9 +219,10 @@ impl OpNode {
     /// use reth_optimism_chainspec::OpChainSpecBuilder;
     /// use reth_optimism_node::OpNode;
     /// use reth_provider::providers::{RocksDBProvider, StaticFileProvider};
+    /// use std::sync::Arc;
     ///
     /// let factory = OpNode::provider_factory_builder()
-    ///     .db(open_db_read_only("db", Default::default()).unwrap())
+    ///     .db(Arc::new(open_db_read_only("db", Default::default()).unwrap()))
     ///     .chainspec(OpChainSpecBuilder::base_mainnet().build().into())
     ///     .static_file(StaticFileProvider::read_only("db/static_files", false).unwrap())
     ///     .rocksdb_provider(RocksDBProvider::builder("db/rocksdb").build().unwrap())

crates/optimism/primitives/src/transaction/signed.rs

@@ -435,11 +435,19 @@ impl reth_codecs::Compact for OpTransactionSigned {
 
         let tx_bits = if zstd_bit {
             let mut tmp = Vec::with_capacity(256);
-            reth_zstd_compressors::with_tx_compressor(|compressor| {
+            if cfg!(feature = "std") {
+                reth_zstd_compressors::TRANSACTION_COMPRESSOR.with(|compressor| {
+                    let mut compressor = compressor.borrow_mut();
+                    let tx_bits = self.transaction.to_compact(&mut tmp);
+                    buf.put_slice(&compressor.compress(&tmp).expect("Failed to compress"));
+                    tx_bits as u8
+                })
+            } else {
+                let mut compressor = reth_zstd_compressors::create_tx_compressor();
                 let tx_bits = self.transaction.to_compact(&mut tmp);
                 buf.put_slice(&compressor.compress(&tmp).expect("Failed to compress"));
                 tx_bits as u8
-            })
+            }
         } else {
             self.transaction.to_compact(buf) as u8
         };
@@ -461,15 +469,29 @@ impl reth_codecs::Compact for OpTransactionSigned {
 
         let zstd_bit = bitflags >> 3;
         let (transaction, buf) = if zstd_bit != 0 {
-            reth_zstd_compressors::with_tx_decompressor(|decompressor| {
-                // TODO: enforce that zstd is only present at a "top" level type
+            if cfg!(feature = "std") {
+                reth_zstd_compressors::TRANSACTION_DECOMPRESSOR.with(|decompressor| {
+                    let mut decompressor = decompressor.borrow_mut();
+
+                    // TODO: enforce that zstd is only present at a "top" level type
+                    let transaction_type = (bitflags & 0b110) >> 1;
+                    let (transaction, _) = OpTypedTransaction::from_compact(
+                        decompressor.decompress(buf),
+                        transaction_type,
+                    );
+
+                    (transaction, buf)
+                })
+            } else {
+                let mut decompressor = reth_zstd_compressors::create_tx_decompressor();
                 let transaction_type = (bitflags & 0b110) >> 1;
                 let (transaction, _) = OpTypedTransaction::from_compact(
                     decompressor.decompress(buf),
                     transaction_type,
                 );
+
                 (transaction, buf)
-            })
+            }
         } else {
             let transaction_type = bitflags >> 1;
             OpTypedTransaction::from_compact(buf, transaction_type)

crates/optimism/rpc/Cargo.toml

@@ -61,7 +61,7 @@ op-revm.workspace = true
 tokio.workspace = true
 futures.workspace = true
 tokio-stream.workspace = true
-reqwest.workspace = true
+reqwest = { workspace = true, features = ["rustls-tls-native-roots"] }
 async-trait.workspace = true
 tower.workspace = true
 

crates/primitives-traits/src/lib.rs

@@ -188,12 +188,6 @@ pub mod serde_bincode_compat;
 pub mod size;
 pub use size::InMemorySize;
 
-/// Rayon utilities
-#[cfg(feature = "rayon")]
-pub mod rayon;
-#[cfg(feature = "rayon")]
-pub use rayon::ParallelBridgeBuffered;
-
 /// Node traits
 pub mod node;
 pub use node::{BlockTy, BodyTy, HeaderTy, NodePrimitives, ReceiptTy, TxTy};

crates/primitives-traits/src/rayon.rs

@@ -1,32 +0,0 @@
-//! Rayon parallel iterator utilities.
-
-use alloc::vec::Vec;
-use rayon::iter::IntoParallelIterator;
-
-/// Extension trait for iterators to convert them to parallel iterators via collection.
-///
-/// This is an alternative to [`rayon::iter::ParallelBridge`] that first collects the iterator
-/// into a `Vec`, then calls [`IntoParallelIterator`] on it. This avoids the mutex contention
-/// that can occur with `par_bridge` when either the iterator's `next()` method is fast or the
-/// parallel tasks are fast, as `par_bridge` wraps the iterator in a mutex.
-///
-/// # When to use
-///
-/// Use `par_bridge_buffered` instead of `par_bridge` when:
-/// - The iterator produces items quickly
-/// - The parallel work per item is relatively light
-/// - The total number of items is known to be reasonable for memory
-///
-/// Stick with `par_bridge` when:
-/// - The iterator is slow (e.g., I/O bound) and you want to overlap iteration with processing
-/// - Memory is constrained and you cannot afford to collect all items upfront
-pub trait ParallelBridgeBuffered: Iterator<Item: Send> + Sized {
-    /// Collects this iterator into a `Vec` and returns a parallel iterator over it.
-    ///
-    /// See [this trait's documentation](ParallelBridgeBuffered) for more details.
-    fn par_bridge_buffered(self) -> rayon::vec::IntoIter<Self::Item> {
-        self.collect::<Vec<_>>().into_par_iter()
-    }
-}
-
-impl<I: Iterator<Item: Send>> ParallelBridgeBuffered for I {}

crates/prune/prune/Cargo.toml

@@ -42,15 +42,11 @@ rayon.workspace = true
 tokio.workspace = true
 rustc-hash.workspace = true
 
-[features]
-rocksdb = ["reth-provider/rocksdb"]
-
 [dev-dependencies]
 # reth
 reth-db = { workspace = true, features = ["test-utils"] }
-reth-stages = { workspace = true, features = ["test-utils", "rocksdb"] }
+reth-stages = { workspace = true, features = ["test-utils"] }
 reth-primitives-traits = { workspace = true, features = ["arbitrary"] }
-reth-storage-api.workspace = true
 reth-testing-utils.workspace = true
 reth-tracing.workspace = true
 

crates/prune/prune/src/builder.rs

@@ -7,10 +7,10 @@ use reth_primitives_traits::NodePrimitives;
 use reth_provider::{
     providers::StaticFileProvider, BlockReader, ChainStateBlockReader, DBProvider,
     DatabaseProviderFactory, NodePrimitivesProvider, PruneCheckpointReader, PruneCheckpointWriter,
-    RocksDBProviderFactory, StageCheckpointReader, StaticFileProviderFactory,
+    StageCheckpointReader, StaticFileProviderFactory, StorageSettingsCache,
 };
 use reth_prune_types::PruneModes;
-use reth_storage_api::{ChangeSetReader, StorageChangeSetReader, StorageSettingsCache};
+use reth_storage_api::{ChangeSetReader, StorageChangeSetReader};
 use std::time::Duration;
 use tokio::sync::watch;
 
@@ -85,7 +85,6 @@ impl PrunerBuilder {
                                 + StageCheckpointReader
                                 + ChangeSetReader
                                 + StorageChangeSetReader
-                                + RocksDBProviderFactory
                                 + StaticFileProviderFactory<
                     Primitives: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>,
                 >,
@@ -122,8 +121,7 @@ impl PrunerBuilder {
             + StorageSettingsCache
             + StageCheckpointReader
             + ChangeSetReader
-            + StorageChangeSetReader
-            + RocksDBProviderFactory,
+            + StorageChangeSetReader,
     {
         let segments = SegmentSet::<Provider>::from_components(static_file_provider, self.segments);
 

crates/prune/prune/src/segments/set.rs

@@ -7,11 +7,10 @@ use reth_db_api::{table::Value, transaction::DbTxMut};
 use reth_primitives_traits::NodePrimitives;
 use reth_provider::{
     providers::StaticFileProvider, BlockReader, ChainStateBlockReader, DBProvider,
-    PruneCheckpointReader, PruneCheckpointWriter, RocksDBProviderFactory,
-    StaticFileProviderFactory,
+    PruneCheckpointReader, PruneCheckpointWriter, StaticFileProviderFactory, StorageSettingsCache,
 };
 use reth_prune_types::PruneModes;
-use reth_storage_api::{ChangeSetReader, StorageChangeSetReader, StorageSettingsCache};
+use reth_storage_api::{ChangeSetReader, StorageChangeSetReader};
 
 /// Collection of [`Segment`]. Thread-safe, allocated on the heap.
 #[derive(Debug)]
@@ -56,8 +55,7 @@ where
         + ChainStateBlockReader
         + StorageSettingsCache
         + ChangeSetReader
-        + StorageChangeSetReader
-        + RocksDBProviderFactory,
+        + StorageChangeSetReader,
 {
     /// Creates a [`SegmentSet`] from an existing components, such as [`StaticFileProvider`] and
     /// [`PruneModes`].

crates/prune/prune/src/segments/user/account_history.rs

@@ -10,20 +10,20 @@ use alloy_primitives::BlockNumber;
 use reth_db_api::{models::ShardedKey, tables, transaction::DbTxMut};
 use reth_provider::{
     changeset_walker::StaticFileAccountChangesetWalker, DBProvider, EitherWriter,
-    RocksDBProviderFactory, StaticFileProviderFactory,
+    StaticFileProviderFactory, StorageSettingsCache,
 };
 use reth_prune_types::{
     PruneMode, PrunePurpose, PruneSegment, SegmentOutput, SegmentOutputCheckpoint,
 };
 use reth_static_file_types::StaticFileSegment;
-use reth_storage_api::{ChangeSetReader, StorageSettingsCache};
+use reth_storage_api::ChangeSetReader;
 use rustc_hash::FxHashMap;
 use tracing::{instrument, trace};
 
 /// Number of account history tables to prune in one step.
 ///
-/// Account History consists of two tables: [`tables::AccountChangeSets`] (either in database or
-/// static files) and [`tables::AccountsHistory`]. We want to prune them to the same block number.
+/// Account History consists of two tables: [`tables::AccountChangeSets`] and
+/// [`tables::AccountsHistory`]. We want to prune them to the same block number.
 const ACCOUNT_HISTORY_TABLES_TO_PRUNE: usize = 2;
 
 #[derive(Debug)]
@@ -42,8 +42,7 @@ where
     Provider: DBProvider<Tx: DbTxMut>
         + StaticFileProviderFactory
         + StorageSettingsCache
-        + ChangeSetReader
-        + RocksDBProviderFactory,
+        + ChangeSetReader,
 {
     fn segment(&self) -> PruneSegment {
         PruneSegment::AccountHistory
@@ -68,13 +67,7 @@ where
         };
         let range_end = *range.end();
 
-        // Check where account history indices are stored
-        #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().account_history_in_rocksdb {
-            return self.prune_rocksdb(provider, input, range, range_end);
-        }
-
-        // Check where account changesets are stored (MDBX path)
+        // Check where account changesets are stored
         if EitherWriter::account_changesets_destination(provider).is_static_file() {
             self.prune_static_files(provider, input, range, range_end)
         } else {
@@ -101,8 +94,6 @@ impl AccountHistory {
             input.limiter
         };
 
-        // The limiter may already be exhausted from a previous segment in the same prune run.
-        // Early exit avoids unnecessary iteration when no budget remains.
         if limiter.is_limit_reached() {
             return Ok(SegmentOutput::not_done(
                 limiter.interrupt_reason(),
@@ -110,14 +101,11 @@ impl AccountHistory {
             ))
         }
 
-        // Deleted account changeset keys (account addresses) with the highest block number deleted
-        // for that key.
-        //
-        // The size of this map is limited by `prune_delete_limit * blocks_since_last_run /
-        // ACCOUNT_HISTORY_TABLES_TO_PRUNE`, and with current default it's usually `3500 * 5
-        // / 2`, so 8750 entries. Each entry is `160 bit + 64 bit`, so the total
-        // size should be up to ~0.25MB + some hashmap overhead. `blocks_since_last_run` is
-        // additionally limited by the `max_reorg_depth`, so no OOM is expected here.
+        // The size of this map it's limited by `prune_delete_limit * blocks_since_last_run /
+        // ACCOUNT_HISTORY_TABLES_TO_PRUNE`, and with the current defaults it's usually `3500 * 5 /
+        // 2`, so 8750 entries. Each entry is `160 bit + 64 bit`, so the total size should be up to
+        // ~0.25MB + some hashmap overhead. `blocks_since_last_run` is additionally limited by the
+        // `max_reorg_depth`, so no OOM is expected here.
         let mut highest_deleted_accounts = FxHashMap::default();
         let mut last_changeset_pruned_block = None;
         let mut pruned_changesets = 0;
@@ -136,8 +124,8 @@ impl AccountHistory {
             limiter.increment_deleted_entries_count();
         }
 
-        // Delete static file jars only when fully processed
-        if done && let Some(last_block) = last_changeset_pruned_block {
+        // Delete static file jars below the pruned block
+        if let Some(last_block) = last_changeset_pruned_block {
             provider
                 .static_file_provider()
                 .delete_segment_below_block(StaticFileSegment::AccountChangeSets, last_block + 1)?;
@@ -222,106 +210,6 @@ impl AccountHistory {
         )
         .map_err(Into::into)
     }
-
-    /// Prunes account history when indices are stored in `RocksDB`.
-    ///
-    /// Reads account changesets from static files and prunes the corresponding
-    /// `RocksDB` history shards.
-    #[cfg(all(unix, feature = "rocksdb"))]
-    fn prune_rocksdb<Provider>(
-        &self,
-        provider: &Provider,
-        input: PruneInput,
-        range: std::ops::RangeInclusive<BlockNumber>,
-        range_end: BlockNumber,
-    ) -> Result<SegmentOutput, PrunerError>
-    where
-        Provider: DBProvider + StaticFileProviderFactory + ChangeSetReader + RocksDBProviderFactory,
-    {
-        use reth_provider::PruneShardOutcome;
-
-        // Unlike MDBX path, we don't divide the limit by 2 because RocksDB path only prunes
-        // history shards (no separate changeset table to delete from). The changesets are in
-        // static files which are deleted separately.
-        let mut limiter = input.limiter;
-
-        if limiter.is_limit_reached() {
-            return Ok(SegmentOutput::not_done(
-                limiter.interrupt_reason(),
-                input.previous_checkpoint.map(SegmentOutputCheckpoint::from_prune_checkpoint),
-            ))
-        }
-
-        let mut highest_deleted_accounts = FxHashMap::default();
-        let mut last_changeset_pruned_block = None;
-        let mut changesets_processed = 0usize;
-        let mut done = true;
-
-        // Walk account changesets from static files using a streaming iterator.
-        // For each changeset, track the highest block number seen for each address
-        // to determine which history shard entries need pruning.
-        let walker = StaticFileAccountChangesetWalker::new(provider, range);
-        for result in walker {
-            if limiter.is_limit_reached() {
-                done = false;
-                break;
-            }
-            let (block_number, changeset) = result?;
-            highest_deleted_accounts.insert(changeset.address, block_number);
-            last_changeset_pruned_block = Some(block_number);
-            changesets_processed += 1;
-            limiter.increment_deleted_entries_count();
-        }
-        trace!(target: "pruner", processed = %changesets_processed, %done, "Scanned account changesets from static files");
-
-        let last_changeset_pruned_block = last_changeset_pruned_block
-            .map(|block_number| if done { block_number } else { block_number.saturating_sub(1) })
-            .unwrap_or(range_end);
-
-        // Prune RocksDB history shards for affected accounts
-        let mut deleted_shards = 0usize;
-        let mut updated_shards = 0usize;
-
-        // Sort by address for better RocksDB cache locality
-        let mut sorted_accounts: Vec<_> = highest_deleted_accounts.into_iter().collect();
-        sorted_accounts.sort_unstable_by_key(|(addr, _)| *addr);
-
-        provider.with_rocksdb_batch(|mut batch| {
-            for (address, highest_block) in &sorted_accounts {
-                let prune_to = (*highest_block).min(last_changeset_pruned_block);
-                match batch.prune_account_history_to(*address, prune_to)? {
-                    PruneShardOutcome::Deleted => deleted_shards += 1,
-                    PruneShardOutcome::Updated => updated_shards += 1,
-                    PruneShardOutcome::Unchanged => {}
-                }
-            }
-            Ok(((), Some(batch.into_inner())))
-        })?;
-        trace!(target: "pruner", deleted = deleted_shards, updated = updated_shards, %done, "Pruned account history (RocksDB indices)");
-
-        // Delete static file jars only when fully processed. During provider.commit(), RocksDB
-        // batch is committed before the MDBX checkpoint. If crash occurs after RocksDB commit
-        // but before MDBX commit, on restart the pruner checkpoint indicates data needs
-        // re-pruning, but the RocksDB shards are already pruned - this is safe because pruning
-        // is idempotent (re-pruning already-pruned shards is a no-op).
-        if done {
-            provider.static_file_provider().delete_segment_below_block(
-                StaticFileSegment::AccountChangeSets,
-                last_changeset_pruned_block + 1,
-            )?;
-        }
-
-        let progress = limiter.progress(done);
-
-        Ok(SegmentOutput {
-            progress,
-            pruned: changesets_processed + deleted_shards + updated_shards,
-            checkpoint: Some(SegmentOutputCheckpoint {
-                block_number: Some(last_changeset_pruned_block),
-                tx_number: None,
-            }),
-        })
-    }
 }
 
 #[cfg(test)]
@@ -651,272 +539,4 @@ mod tests {
         test_prune(998, 2, (PruneProgress::Finished, 1000));
         test_prune(1400, 3, (PruneProgress::Finished, 804));
     }
-
-    #[cfg(all(unix, feature = "rocksdb"))]
-    #[test]
-    fn prune_rocksdb_path() {
-        use reth_db_api::models::ShardedKey;
-        use reth_provider::{RocksDBProviderFactory, StaticFileProviderFactory};
-
-        let db = TestStageDB::default();
-        let mut rng = generators::rng();
-
-        let blocks = random_block_range(
-            &mut rng,
-            0..=100,
-            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
-        );
-        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
-
-        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
-
-        let (changesets, _) = random_changeset_range(
-            &mut rng,
-            blocks.iter(),
-            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
-            0..0,
-            0..0,
-        );
-
-        db.insert_changesets_to_static_files(changesets.clone(), None)
-            .expect("insert changesets to static files");
-
-        let mut account_blocks: BTreeMap<_, Vec<u64>> = BTreeMap::new();
-        for (block, changeset) in changesets.iter().enumerate() {
-            for (address, _, _) in changeset {
-                account_blocks.entry(*address).or_default().push(block as u64);
-            }
-        }
-
-        let rocksdb = db.factory.rocksdb_provider();
-        let mut batch = rocksdb.batch();
-        for (address, block_numbers) in &account_blocks {
-            let shard = BlockNumberList::new_pre_sorted(block_numbers.iter().copied());
-            batch
-                .put::<tables::AccountsHistory>(ShardedKey::new(*address, u64::MAX), &shard)
-                .unwrap();
-        }
-        batch.commit().unwrap();
-
-        for (address, expected_blocks) in &account_blocks {
-            let shards = rocksdb.account_history_shards(*address).unwrap();
-            assert_eq!(shards.len(), 1);
-            assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), *expected_blocks);
-        }
-
-        let to_block: BlockNumber = 50;
-        let prune_mode = PruneMode::Before(to_block);
-        let input =
-            PruneInput { previous_checkpoint: None, to_block, limiter: PruneLimiter::default() };
-        let segment = AccountHistory::new(prune_mode);
-
-        db.factory.set_storage_settings_cache(
-            StorageSettings::default()
-                .with_account_changesets_in_static_files(true)
-                .with_account_history_in_rocksdb(true),
-        );
-
-        let provider = db.factory.database_provider_rw().unwrap();
-        let result = segment.prune(&provider, input).unwrap();
-        provider.commit().expect("commit");
-
-        assert_matches!(
-            result,
-            SegmentOutput { progress: PruneProgress::Finished, pruned, checkpoint: Some(_) }
-                if pruned > 0
-        );
-
-        for (address, original_blocks) in &account_blocks {
-            let shards = rocksdb.account_history_shards(*address).unwrap();
-
-            let expected_blocks: Vec<u64> =
-                original_blocks.iter().copied().filter(|b| *b > to_block).collect();
-
-            if expected_blocks.is_empty() {
-                assert!(
-                    shards.is_empty(),
-                    "Expected no shards for address {address:?} after pruning"
-                );
-            } else {
-                assert_eq!(shards.len(), 1, "Expected 1 shard for address {address:?}");
-                assert_eq!(
-                    shards[0].1.iter().collect::<Vec<_>>(),
-                    expected_blocks,
-                    "Shard blocks mismatch for address {address:?}"
-                );
-            }
-        }
-
-        let static_file_provider = db.factory.static_file_provider();
-        let highest_block = static_file_provider.get_highest_static_file_block(
-            reth_static_file_types::StaticFileSegment::AccountChangeSets,
-        );
-        if let Some(block) = highest_block {
-            assert!(
-                block > to_block,
-                "Static files should only contain blocks above to_block ({to_block}), got {block}"
-            );
-        }
-    }
-
-    /// Tests that when a limiter stops mid-block (with multiple changes for the same block),
-    /// the checkpoint is set to `block_number - 1` to avoid dangling index entries.
-    #[test]
-    fn prune_partial_progress_mid_block() {
-        use alloy_primitives::{Address, U256};
-        use reth_primitives_traits::Account;
-        use reth_testing_utils::generators::ChangeSet;
-
-        let db = TestStageDB::default();
-        let mut rng = generators::rng();
-
-        // Create blocks 0..=10
-        let blocks = random_block_range(
-            &mut rng,
-            0..=10,
-            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
-        );
-        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
-
-        // Create specific changesets where block 5 has 4 account changes
-        let addr1 = Address::with_last_byte(1);
-        let addr2 = Address::with_last_byte(2);
-        let addr3 = Address::with_last_byte(3);
-        let addr4 = Address::with_last_byte(4);
-        let addr5 = Address::with_last_byte(5);
-
-        let account = Account { nonce: 1, balance: U256::from(100), bytecode_hash: None };
-
-        // Build changesets: blocks 0-4 have 1 change each, block 5 has 4 changes, block 6 has 1
-        let changesets: Vec<ChangeSet> = vec![
-            vec![(addr1, account, vec![])], // block 0
-            vec![(addr1, account, vec![])], // block 1
-            vec![(addr1, account, vec![])], // block 2
-            vec![(addr1, account, vec![])], // block 3
-            vec![(addr1, account, vec![])], // block 4
-            // block 5: 4 different account changes (sorted by address for consistency)
-            vec![
-                (addr1, account, vec![]),
-                (addr2, account, vec![]),
-                (addr3, account, vec![]),
-                (addr4, account, vec![]),
-            ],
-            vec![(addr5, account, vec![])], // block 6
-        ];
-
-        db.insert_changesets(changesets.clone(), None).expect("insert changesets");
-        db.insert_history(changesets.clone(), None).expect("insert history");
-
-        // Total changesets: 5 (blocks 0-4) + 4 (block 5) + 1 (block 6) = 10
-        assert_eq!(
-            db.table::<tables::AccountChangeSets>().unwrap().len(),
-            changesets.iter().flatten().count()
-        );
-
-        let prune_mode = PruneMode::Before(10);
-
-        // Set limiter to stop after 7 entries (mid-block 5: 5 from blocks 0-4, then 2 of 4 from
-        // block 5). Due to ACCOUNT_HISTORY_TABLES_TO_PRUNE=2, actual limit is 7/2=3
-        // changesets. So we'll process blocks 0, 1, 2 (3 changesets), stopping before block
-        // 3. Actually, let's use a higher limit to reach block 5. With limit=14, we get 7
-        // changeset slots. Blocks 0-4 use 5 slots, leaving 2 for block 5 (which has 4), so
-        // we stop mid-block 5.
-        let deleted_entries_limit = 14; // 14/2 = 7 changeset entries before limit
-        let limiter = PruneLimiter::default().set_deleted_entries_limit(deleted_entries_limit);
-
-        let input = PruneInput { previous_checkpoint: None, to_block: 10, limiter };
-        let segment = AccountHistory::new(prune_mode);
-
-        let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(
-            StorageSettings::default().with_account_changesets_in_static_files(false),
-        );
-        let result = segment.prune(&provider, input).unwrap();
-
-        // Should report that there's more data
-        assert!(!result.progress.is_finished(), "Expected HasMoreData since we stopped mid-block");
-
-        // Save checkpoint and commit
-        segment
-            .save_checkpoint(&provider, result.checkpoint.unwrap().as_prune_checkpoint(prune_mode))
-            .unwrap();
-        provider.commit().expect("commit");
-
-        // Verify checkpoint is set to block 4 (not 5), since block 5 is incomplete
-        let checkpoint = db
-            .factory
-            .provider()
-            .unwrap()
-            .get_prune_checkpoint(PruneSegment::AccountHistory)
-            .unwrap()
-            .expect("checkpoint should exist");
-
-        assert_eq!(
-            checkpoint.block_number,
-            Some(4),
-            "Checkpoint should be block 4 (block before incomplete block 5)"
-        );
-
-        // Verify remaining changesets (block 5 and 6 should still have entries)
-        let remaining_changesets = db.table::<tables::AccountChangeSets>().unwrap();
-        // After pruning blocks 0-4, remaining should be block 5 (4 entries) + block 6 (1 entry) = 5
-        // But since we stopped mid-block 5, some of block 5 might be pruned
-        // However, checkpoint is 4, so on re-run we should re-process from block 5
-        assert!(
-            !remaining_changesets.is_empty(),
-            "Should have remaining changesets for blocks 5-6"
-        );
-
-        // Verify no dangling history indices for blocks that weren't fully pruned
-        // The indices for block 5 should still reference blocks <= 5 appropriately
-        let history = db.table::<tables::AccountsHistory>().unwrap();
-        for (key, _blocks) in &history {
-            // All blocks in the history should be > checkpoint block number
-            // OR the shard's highest_block_number should be > checkpoint
-            assert!(
-                key.highest_block_number > 4,
-                "Found stale history shard with highest_block_number {} <= checkpoint 4",
-                key.highest_block_number
-            );
-        }
-
-        // Run prune again to complete - should finish processing block 5 and 6
-        let input2 = PruneInput {
-            previous_checkpoint: Some(checkpoint),
-            to_block: 10,
-            limiter: PruneLimiter::default().set_deleted_entries_limit(100), // high limit
-        };
-
-        let provider2 = db.factory.database_provider_rw().unwrap();
-        provider2.set_storage_settings_cache(
-            StorageSettings::default().with_account_changesets_in_static_files(false),
-        );
-        let result2 = segment.prune(&provider2, input2).unwrap();
-
-        assert!(result2.progress.is_finished(), "Second run should complete");
-
-        segment
-            .save_checkpoint(
-                &provider2,
-                result2.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
-            )
-            .unwrap();
-        provider2.commit().expect("commit");
-
-        // Verify final checkpoint
-        let final_checkpoint = db
-            .factory
-            .provider()
-            .unwrap()
-            .get_prune_checkpoint(PruneSegment::AccountHistory)
-            .unwrap()
-            .expect("checkpoint should exist");
-
-        // Should now be at block 6 (the last block with changesets)
-        assert_eq!(final_checkpoint.block_number, Some(6), "Final checkpoint should be at block 6");
-
-        // All changesets should be pruned
-        let final_changesets = db.table::<tables::AccountChangeSets>().unwrap();
-        assert!(final_changesets.is_empty(), "All changesets up to block 10 should be pruned");
-    }
 }

crates/prune/prune/src/segments/user/sender_recovery.rs

@@ -1,18 +1,14 @@
 use crate::{
     db_ext::DbTxPruneExt,
-    segments::{self, PruneInput, Segment},
+    segments::{PruneInput, Segment},
     PrunerError,
 };
 use reth_db_api::{tables, transaction::DbTxMut};
-use reth_provider::{
-    BlockReader, DBProvider, EitherWriterDestination, StaticFileProviderFactory,
-    StorageSettingsCache, TransactionsProvider,
-};
+use reth_provider::{BlockReader, DBProvider, TransactionsProvider};
 use reth_prune_types::{
     PruneMode, PruneProgress, PrunePurpose, PruneSegment, SegmentOutput, SegmentOutputCheckpoint,
 };
-use reth_static_file_types::StaticFileSegment;
-use tracing::{debug, instrument, trace};
+use tracing::{instrument, trace};
 
 #[derive(Debug)]
 pub struct SenderRecovery {
@@ -27,11 +23,7 @@ impl SenderRecovery {
 
 impl<Provider> Segment<Provider> for SenderRecovery
 where
-    Provider: DBProvider<Tx: DbTxMut>
-        + TransactionsProvider
-        + BlockReader
-        + StorageSettingsCache
-        + StaticFileProviderFactory,
+    Provider: DBProvider<Tx: DbTxMut> + TransactionsProvider + BlockReader,
 {
     fn segment(&self) -> PruneSegment {
         PruneSegment::SenderRecovery
@@ -47,16 +39,6 @@ where
 
     #[instrument(target = "pruner", skip(self, provider), ret(level = "trace"))]
     fn prune(&self, provider: &Provider, input: PruneInput) -> Result<SegmentOutput, PrunerError> {
-        if EitherWriterDestination::senders(provider).is_static_file() {
-            debug!(target: "pruner", "Pruning transaction senders from static files.");
-            return segments::prune_static_files(
-                provider,
-                input,
-                StaticFileSegment::TransactionSenders,
-            )
-        }
-        debug!(target: "pruner", "Pruning transaction senders from database.");
-
         let tx_range = match input.get_next_tx_num_range(provider)? {
             Some(range) => range,
             None => {

crates/prune/prune/src/segments/user/storage_history.rs

@@ -12,7 +12,7 @@ use reth_db_api::{
     tables,
     transaction::DbTxMut,
 };
-use reth_provider::{DBProvider, EitherWriter, RocksDBProviderFactory, StaticFileProviderFactory};
+use reth_provider::{DBProvider, EitherWriter, StaticFileProviderFactory};
 use reth_prune_types::{
     PruneMode, PrunePurpose, PruneSegment, SegmentOutput, SegmentOutputCheckpoint,
 };
@@ -43,8 +43,7 @@ where
     Provider: DBProvider<Tx: DbTxMut>
         + StaticFileProviderFactory
         + StorageChangeSetReader
-        + StorageSettingsCache
-        + RocksDBProviderFactory,
+        + StorageSettingsCache,
 {
     fn segment(&self) -> PruneSegment {
         PruneSegment::StorageHistory
@@ -69,13 +68,6 @@ where
         };
         let range_end = *range.end();
 
-        // Check where storage history indices are stored
-        #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().storages_history_in_rocksdb {
-            return self.prune_rocksdb(provider, input, range, range_end);
-        }
-
-        // Check where storage changesets are stored (MDBX path)
         if EitherWriter::storage_changesets_destination(provider).is_static_file() {
             self.prune_static_files(provider, input, range, range_end)
         } else {
@@ -102,8 +94,6 @@ impl StorageHistory {
             input.limiter
         };
 
-        // The limiter may already be exhausted from a previous segment in the same prune run.
-        // Early exit avoids unnecessary iteration when no budget remains.
         if limiter.is_limit_reached() {
             return Ok(SegmentOutput::not_done(
                 limiter.interrupt_reason(),
@@ -136,8 +126,8 @@ impl StorageHistory {
             limiter.increment_deleted_entries_count();
         }
 
-        // Delete static file jars only when fully processed
-        if done && let Some(last_block) = last_changeset_pruned_block {
+        // Delete static file jars below the pruned block
+        if let Some(last_block) = last_changeset_pruned_block {
             provider
                 .static_file_provider()
                 .delete_segment_below_block(StaticFileSegment::StorageChangeSets, last_block + 1)?;
@@ -226,107 +216,6 @@ impl StorageHistory {
         )
         .map_err(Into::into)
     }
-
-    /// Prunes storage history when indices are stored in `RocksDB`.
-    ///
-    /// Reads storage changesets from static files and prunes the corresponding
-    /// `RocksDB` history shards.
-    #[cfg(all(unix, feature = "rocksdb"))]
-    fn prune_rocksdb<Provider>(
-        &self,
-        provider: &Provider,
-        input: PruneInput,
-        range: std::ops::RangeInclusive<BlockNumber>,
-        range_end: BlockNumber,
-    ) -> Result<SegmentOutput, PrunerError>
-    where
-        Provider: DBProvider + StaticFileProviderFactory + RocksDBProviderFactory,
-    {
-        use reth_provider::PruneShardOutcome;
-
-        let mut limiter = input.limiter;
-
-        if limiter.is_limit_reached() {
-            return Ok(SegmentOutput::not_done(
-                limiter.interrupt_reason(),
-                input.previous_checkpoint.map(SegmentOutputCheckpoint::from_prune_checkpoint),
-            ))
-        }
-
-        let mut highest_deleted_storages: FxHashMap<_, _> = FxHashMap::default();
-        let mut last_changeset_pruned_block = None;
-        let mut changesets_processed = 0usize;
-        let mut done = true;
-
-        // Walk storage changesets from static files using a streaming iterator.
-        // For each changeset, track the highest block number seen for each (address, storage_key)
-        // pair to determine which history shard entries need pruning.
-        let walker = provider.static_file_provider().walk_storage_changeset_range(range);
-        for result in walker {
-            if limiter.is_limit_reached() {
-                done = false;
-                break;
-            }
-            let (block_address, entry) = result?;
-            let block_number = block_address.block_number();
-            let address = block_address.address();
-            highest_deleted_storages.insert((address, entry.key), block_number);
-            last_changeset_pruned_block = Some(block_number);
-            changesets_processed += 1;
-            limiter.increment_deleted_entries_count();
-        }
-
-        trace!(target: "pruner", processed = %changesets_processed, %done, "Scanned storage changesets from static files");
-
-        let last_changeset_pruned_block = last_changeset_pruned_block
-            .map(|block_number| if done { block_number } else { block_number.saturating_sub(1) })
-            .unwrap_or(range_end);
-
-        // Prune RocksDB history shards for affected storage slots
-        let mut deleted_shards = 0usize;
-        let mut updated_shards = 0usize;
-
-        // Sort by (address, storage_key) for better RocksDB cache locality
-        let mut sorted_storages: Vec<_> = highest_deleted_storages.into_iter().collect();
-        sorted_storages.sort_unstable_by_key(|((addr, key), _)| (*addr, *key));
-
-        provider.with_rocksdb_batch(|mut batch| {
-            for ((address, storage_key), highest_block) in &sorted_storages {
-                let prune_to = (*highest_block).min(last_changeset_pruned_block);
-                match batch.prune_storage_history_to(*address, *storage_key, prune_to)? {
-                    PruneShardOutcome::Deleted => deleted_shards += 1,
-                    PruneShardOutcome::Updated => updated_shards += 1,
-                    PruneShardOutcome::Unchanged => {}
-                }
-            }
-            Ok(((), Some(batch.into_inner())))
-        })?;
-
-        trace!(target: "pruner", deleted = deleted_shards, updated = updated_shards, %done, "Pruned storage history (RocksDB indices)");
-
-        // Delete static file jars only when fully processed. During provider.commit(), RocksDB
-        // batch is committed before the MDBX checkpoint. If crash occurs after RocksDB commit
-        // but before MDBX commit, on restart the pruner checkpoint indicates data needs
-        // re-pruning, but the RocksDB shards are already pruned - this is safe because pruning
-        // is idempotent (re-pruning already-pruned shards is a no-op).
-        if done {
-            provider.static_file_provider().delete_segment_below_block(
-                StaticFileSegment::StorageChangeSets,
-                last_changeset_pruned_block + 1,
-            )?;
-        }
-
-        let progress = limiter.progress(done);
-
-        Ok(SegmentOutput {
-            progress,
-            pruned: changesets_processed + deleted_shards + updated_shards,
-            checkpoint: Some(SegmentOutputCheckpoint {
-                block_number: Some(last_changeset_pruned_block),
-                tx_number: None,
-            }),
-        })
-    }
 }
 
 #[cfg(test)]
@@ -664,270 +553,4 @@ mod tests {
         test_prune(998, 2, (PruneProgress::Finished, 500));
         test_prune(1200, 3, (PruneProgress::Finished, 202));
     }
-
-    /// Tests that when a limiter stops mid-block (with multiple storage changes for the same
-    /// block), the checkpoint is set to `block_number - 1` to avoid dangling index entries.
-    #[test]
-    fn prune_partial_progress_mid_block() {
-        use alloy_primitives::{Address, U256};
-        use reth_primitives_traits::Account;
-        use reth_testing_utils::generators::ChangeSet;
-
-        let db = TestStageDB::default();
-        let mut rng = generators::rng();
-
-        // Create blocks 0..=10
-        let blocks = random_block_range(
-            &mut rng,
-            0..=10,
-            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
-        );
-        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
-
-        // Create specific changesets where block 5 has 4 storage changes
-        let addr1 = Address::with_last_byte(1);
-        let addr2 = Address::with_last_byte(2);
-
-        let account = Account { nonce: 1, balance: U256::from(100), bytecode_hash: None };
-
-        // Create storage entries
-        let storage_entry = |key: u8| reth_primitives_traits::StorageEntry {
-            key: B256::with_last_byte(key),
-            value: U256::from(100),
-        };
-
-        // Build changesets: blocks 0-4 have 1 storage change each, block 5 has 4 changes, block 6
-        // has 1. Entries within each account must be sorted by key.
-        let changesets: Vec<ChangeSet> = vec![
-            vec![(addr1, account, vec![storage_entry(1)])], // block 0
-            vec![(addr1, account, vec![storage_entry(1)])], // block 1
-            vec![(addr1, account, vec![storage_entry(1)])], // block 2
-            vec![(addr1, account, vec![storage_entry(1)])], // block 3
-            vec![(addr1, account, vec![storage_entry(1)])], // block 4
-            // block 5: 4 different storage changes (2 addresses, each with 2 storage slots)
-            // Sorted by address, then by storage key within each address
-            vec![
-                (addr1, account, vec![storage_entry(1), storage_entry(2)]),
-                (addr2, account, vec![storage_entry(1), storage_entry(2)]),
-            ],
-            vec![(addr1, account, vec![storage_entry(3)])], // block 6
-        ];
-
-        db.insert_changesets(changesets.clone(), None).expect("insert changesets");
-        db.insert_history(changesets.clone(), None).expect("insert history");
-
-        // Total storage changesets
-        let total_storage_entries: usize =
-            changesets.iter().flat_map(|c| c.iter()).map(|(_, _, entries)| entries.len()).sum();
-        assert_eq!(db.table::<tables::StorageChangeSets>().unwrap().len(), total_storage_entries);
-
-        let prune_mode = PruneMode::Before(10);
-
-        // Set limiter to stop mid-block 5
-        // With STORAGE_HISTORY_TABLES_TO_PRUNE=2, limit=14 gives us 7 storage entries before limit
-        // Blocks 0-4 use 5 slots, leaving 2 for block 5 (which has 4), so we stop mid-block 5
-        let deleted_entries_limit = 14; // 14/2 = 7 storage entries before limit
-        let limiter = PruneLimiter::default().set_deleted_entries_limit(deleted_entries_limit);
-
-        let input = PruneInput { previous_checkpoint: None, to_block: 10, limiter };
-        let segment = StorageHistory::new(prune_mode);
-
-        let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(
-            StorageSettings::default().with_storage_changesets_in_static_files(false),
-        );
-        let result = segment.prune(&provider, input).unwrap();
-
-        // Should report that there's more data
-        assert!(!result.progress.is_finished(), "Expected HasMoreData since we stopped mid-block");
-
-        // Save checkpoint and commit
-        segment
-            .save_checkpoint(&provider, result.checkpoint.unwrap().as_prune_checkpoint(prune_mode))
-            .unwrap();
-        provider.commit().expect("commit");
-
-        // Verify checkpoint is set to block 4 (not 5), since block 5 is incomplete
-        let checkpoint = db
-            .factory
-            .provider()
-            .unwrap()
-            .get_prune_checkpoint(PruneSegment::StorageHistory)
-            .unwrap()
-            .expect("checkpoint should exist");
-
-        assert_eq!(
-            checkpoint.block_number,
-            Some(4),
-            "Checkpoint should be block 4 (block before incomplete block 5)"
-        );
-
-        // Verify remaining changesets
-        let remaining_changesets = db.table::<tables::StorageChangeSets>().unwrap();
-        assert!(
-            !remaining_changesets.is_empty(),
-            "Should have remaining changesets for blocks 5-6"
-        );
-
-        // Verify no dangling history indices for blocks that weren't fully pruned
-        let history = db.table::<tables::StoragesHistory>().unwrap();
-        for (key, _blocks) in &history {
-            assert!(
-                key.sharded_key.highest_block_number > 4,
-                "Found stale history shard with highest_block_number {} <= checkpoint 4",
-                key.sharded_key.highest_block_number
-            );
-        }
-
-        // Run prune again to complete - should finish processing block 5 and 6
-        let input2 = PruneInput {
-            previous_checkpoint: Some(checkpoint),
-            to_block: 10,
-            limiter: PruneLimiter::default().set_deleted_entries_limit(100), // high limit
-        };
-
-        let provider2 = db.factory.database_provider_rw().unwrap();
-        provider2.set_storage_settings_cache(
-            StorageSettings::default().with_storage_changesets_in_static_files(false),
-        );
-        let result2 = segment.prune(&provider2, input2).unwrap();
-
-        assert!(result2.progress.is_finished(), "Second run should complete");
-
-        segment
-            .save_checkpoint(
-                &provider2,
-                result2.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
-            )
-            .unwrap();
-        provider2.commit().expect("commit");
-
-        // Verify final checkpoint
-        let final_checkpoint = db
-            .factory
-            .provider()
-            .unwrap()
-            .get_prune_checkpoint(PruneSegment::StorageHistory)
-            .unwrap()
-            .expect("checkpoint should exist");
-
-        // Should now be at block 6 (the last block with changesets)
-        assert_eq!(final_checkpoint.block_number, Some(6), "Final checkpoint should be at block 6");
-
-        // All changesets should be pruned
-        let final_changesets = db.table::<tables::StorageChangeSets>().unwrap();
-        assert!(final_changesets.is_empty(), "All changesets up to block 10 should be pruned");
-    }
-
-    #[cfg(all(unix, feature = "rocksdb"))]
-    #[test]
-    fn prune_rocksdb() {
-        use reth_db_api::models::storage_sharded_key::StorageShardedKey;
-        use reth_provider::RocksDBProviderFactory;
-        use reth_storage_api::StorageSettings;
-
-        let db = TestStageDB::default();
-        let mut rng = generators::rng();
-
-        let blocks = random_block_range(
-            &mut rng,
-            0..=100,
-            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
-        );
-        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
-
-        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
-
-        let (changesets, _) = random_changeset_range(
-            &mut rng,
-            blocks.iter(),
-            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
-            1..2,
-            1..2,
-        );
-
-        db.insert_changesets_to_static_files(changesets.clone(), None)
-            .expect("insert changesets to static files");
-
-        let mut storage_indices: BTreeMap<(alloy_primitives::Address, B256), Vec<u64>> =
-            BTreeMap::new();
-        for (block, changeset) in changesets.iter().enumerate() {
-            for (address, _, storage_entries) in changeset {
-                for entry in storage_entries {
-                    storage_indices.entry((*address, entry.key)).or_default().push(block as u64);
-                }
-            }
-        }
-
-        {
-            let rocksdb = db.factory.rocksdb_provider();
-            let mut batch = rocksdb.batch();
-            for ((address, storage_key), block_numbers) in &storage_indices {
-                let shard = BlockNumberList::new_pre_sorted(block_numbers.clone());
-                batch
-                    .put::<tables::StoragesHistory>(
-                        StorageShardedKey::last(*address, *storage_key),
-                        &shard,
-                    )
-                    .expect("insert storage history shard");
-            }
-            batch.commit().expect("commit rocksdb batch");
-        }
-
-        {
-            let rocksdb = db.factory.rocksdb_provider();
-            for (address, storage_key) in storage_indices.keys() {
-                let shards = rocksdb.storage_history_shards(*address, *storage_key).unwrap();
-                assert!(!shards.is_empty(), "RocksDB should contain storage history before prune");
-            }
-        }
-
-        let to_block = 50u64;
-        let prune_mode = PruneMode::Before(to_block);
-        let input =
-            PruneInput { previous_checkpoint: None, to_block, limiter: PruneLimiter::default() };
-        let segment = StorageHistory::new(prune_mode);
-
-        let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(
-            StorageSettings::default()
-                .with_storage_changesets_in_static_files(true)
-                .with_storages_history_in_rocksdb(true),
-        );
-        let result = segment.prune(&provider, input).unwrap();
-        provider.commit().expect("commit");
-
-        assert_matches!(
-            result,
-            SegmentOutput { progress: PruneProgress::Finished, checkpoint: Some(_), .. }
-        );
-
-        {
-            let rocksdb = db.factory.rocksdb_provider();
-            for ((address, storage_key), block_numbers) in &storage_indices {
-                let shards = rocksdb.storage_history_shards(*address, *storage_key).unwrap();
-
-                let remaining_blocks: Vec<u64> =
-                    block_numbers.iter().copied().filter(|&b| b > to_block).collect();
-
-                if remaining_blocks.is_empty() {
-                    assert!(
-                        shards.is_empty(),
-                        "Shard for {:?}/{:?} should be deleted when all blocks pruned",
-                        address,
-                        storage_key
-                    );
-                } else {
-                    assert!(!shards.is_empty(), "Shard should exist with remaining blocks");
-                    let actual_blocks: Vec<u64> =
-                        shards.iter().flat_map(|(_, list)| list.iter()).collect();
-                    assert_eq!(
-                        actual_blocks, remaining_blocks,
-                        "RocksDB shard should only contain blocks > {}",
-                        to_block
-                    );
-                }
-            }
-        }
-    }
 }

crates/prune/types/src/target.rs

@@ -54,7 +54,7 @@ pub struct PruneModes {
     pub transaction_lookup: Option<PruneMode>,
     /// Receipts pruning configuration. This setting overrides `receipts_log_filter`
     /// and offers improved performance.
-    #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none"))]
+    #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none",))]
     pub receipts: Option<PruneMode>,
     /// Account History pruning configuration.
     #[cfg_attr(
@@ -75,7 +75,7 @@ pub struct PruneModes {
     )]
     pub storage_history: Option<PruneMode>,
     /// Bodies History pruning configuration.
-    #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none"))]
+    #[cfg_attr(any(test, feature = "serde"), serde(skip_serializing_if = "Option::is_none",))]
     pub bodies_history: Option<PruneMode>,
     /// Receipts pruning configuration by retaining only those receipts that contain logs emitted
     /// by the specified addresses, discarding others. This setting is overridden by `receipts`.
@@ -112,13 +112,7 @@ impl PruneModes {
     ///
     /// Returns `true` if any migration was performed.
     pub const fn migrate(&mut self) -> bool {
-        match &self.receipts {
-            Some(PruneMode::Full | PruneMode::Distance(0..MINIMUM_DISTANCE)) => {
-                self.receipts = Some(PruneMode::Distance(MINIMUM_DISTANCE));
-                true
-            }
-            _ => false,
-        }
+        false
     }
 
     /// Returns an error if we can't unwind to the targeted block because the target block is

crates/rpc/rpc-api/src/debug.rs

@@ -2,7 +2,7 @@ use alloy_eip7928::BlockAccessList;
 use alloy_eips::{BlockId, BlockNumberOrTag};
 use alloy_genesis::ChainConfig;
 use alloy_json_rpc::RpcObject;
-use alloy_primitives::{Address, Bytes, B256, U64};
+use alloy_primitives::{Address, Bytes, B256};
 use alloy_rpc_types_debug::ExecutionWitness;
 use alloy_rpc_types_eth::{Bundle, StateContext};
 use alloy_rpc_types_trace::geth::{
@@ -325,7 +325,7 @@ pub trait DebugApi<TxReq: RpcObject> {
     /// Sets the current head of the local chain by block number. Note, this is a destructive action
     /// and may severely damage your chain. Use with extreme caution.
     #[method(name = "setHead")]
-    async fn debug_set_head(&self, number: U64) -> RpcResult<()>;
+    async fn debug_set_head(&self, number: u64) -> RpcResult<()>;
 
     /// Sets the rate of mutex profiling.
     #[method(name = "setMutexProfileFraction")]

crates/rpc/rpc-builder/src/config.rs

@@ -107,7 +107,6 @@ impl RethRpcServerConfig for RpcServerArgs {
             .pending_block_kind(self.rpc_pending_block)
             .raw_tx_forwarder(self.rpc_forwarder.clone())
             .rpc_evm_memory_limit(self.rpc_evm_memory_limit)
-            .force_blob_sidecar_upcasting(self.rpc_force_blob_sidecar_upcasting)
     }
 
     fn flashbots_config(&self) -> ValidationApiConfig {

crates/rpc/rpc-eth-types/Cargo.toml

@@ -49,7 +49,7 @@ jsonrpsee-types.workspace = true
 futures.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
-reqwest.workspace = true
+reqwest = { workspace = true, features = ["rustls-tls-native-roots"] }
 
 # metrics
 metrics.workspace = true

crates/rpc/rpc-eth-types/src/builder/config.rs

@@ -107,11 +107,6 @@ pub struct EthConfig {
     pub send_raw_transaction_sync_timeout: Duration,
     /// Maximum memory the EVM can allocate per RPC request.
     pub rpc_evm_memory_limit: u64,
-    /// Whether to force upcasting EIP-4844 blob sidecars to EIP-7594 format when Osaka is active.
-    ///
-    /// This is disabled by default, allowing blob transactions with EIP-4844 sidecars to be
-    /// submitted without automatic conversion.
-    pub force_blob_sidecar_upcasting: bool,
 }
 
 impl EthConfig {
@@ -145,7 +140,6 @@ impl Default for EthConfig {
             raw_tx_forwarder: ForwardConfig::default(),
             send_raw_transaction_sync_timeout: RPC_DEFAULT_SEND_RAW_TX_SYNC_TIMEOUT_SECS,
             rpc_evm_memory_limit: (1 << 32) - 1,
-            force_blob_sidecar_upcasting: false,
         }
     }
 }
@@ -248,12 +242,6 @@ impl EthConfig {
         self.rpc_evm_memory_limit = memory_limit;
         self
     }
-
-    /// Configures whether to force upcasting EIP-4844 blob sidecars to EIP-7594 format.
-    pub const fn force_blob_sidecar_upcasting(mut self, force: bool) -> Self {
-        self.force_blob_sidecar_upcasting = force;
-        self
-    }
 }
 
 /// Config for the filter

crates/rpc/rpc/Cargo.toml

@@ -71,6 +71,10 @@ revm-primitives = { workspace = true, features = ["serde"] }
 
 # rpc
 jsonrpsee.workspace = true
+http.workspace = true
+http-body.workspace = true
+hyper.workspace = true
+jsonwebtoken.workspace = true
 serde_json.workspace = true
 jsonrpsee-types.workspace = true
 
@@ -78,6 +82,7 @@ jsonrpsee-types.workspace = true
 async-trait.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream.workspace = true
+tower.workspace = true
 pin-project.workspace = true
 parking_lot.workspace = true